This patch moves tools/gatt-example to test/ and renames it to
example-gatt-server in preparation for example-gatt-client.
The same naming convention (example-*) is also adopted for
test/advertisement-example, so it's now called
test/example-advertisement. Also, both scripts now have the execute
set appropriately to be consistent with the other Python scripts.
---
test/advertisement-example | 170 ---------------
test/example-advertisement | 170 +++++++++++++++
test/example-gatt-server | 533 +++++++++++++++++++++++++++++++++++++++++++++
tools/gatt-example | 533 ---------------------------------------------
4 files changed, 703 insertions(+), 703 deletions(-)
delete mode 100644 test/advertisement-example
create mode 100755 test/example-advertisement
create mode 100755 test/example-gatt-server
delete mode 100644 tools/gatt-example
diff --git a/test/advertisement-example b/test/advertisement-example
deleted file mode 100644
index 98aeafa..0000000
--- a/test/advertisement-example
+++ /dev/null
@@ -1,170 +0,0 @@
-#!/usr/bin/python
-
-import dbus
-import dbus.exceptions
-import dbus.mainloop.glib
-import dbus.service
-
-import array
-import gobject
-
-from random import randint
-
-mainloop = None
-
-BLUEZ_SERVICE_NAME = 'org.bluez'
-LE_ADVERTISING_MANAGER_IFACE = 'org.bluez.LEAdvertisingManager1'
-DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
-DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
-
-LE_ADVERTISEMENT_IFACE = 'org.bluez.LEAdvertisement1'
-
-
-class InvalidArgsException(dbus.exceptions.DBusException):
- _dbus_error_name = 'org.freedesktop.DBus.Error.InvalidArgs'
-
-
-class NotSupportedException(dbus.exceptions.DBusException):
- _dbus_error_name = 'org.bluez.Error.NotSupported'
-
-
-class NotPermittedException(dbus.exceptions.DBusException):
- _dbus_error_name = 'org.bluez.Error.NotPermitted'
-
-
-class InvalidValueLengthException(dbus.exceptions.DBusException):
- _dbus_error_name = 'org.bluez.Error.InvalidValueLength'
-
-
-class FailedException(dbus.exceptions.DBusException):
- _dbus_error_name = 'org.bluez.Error.Failed'
-
-
-class Advertisement(dbus.service.Object):
- PATH_BASE = '/org/bluez/example/advertisement'
-
- def __init__(self, bus, index, advertising_type):
- self.path = self.PATH_BASE + str(index)
- self.bus = bus
- self.ad_type = advertising_type
- self.service_uuids = None
- self.manufacturer_data = None
- self.solicit_uuids = None
- self.service_data = None
- dbus.service.Object.__init__(self, bus, self.path)
-
- def get_properties(self):
- properties = dict()
- properties['Type'] = self.ad_type
- if self.service_uuids is not None:
- properties['ServiceUUIDs'] = dbus.Array(self.service_uuids,
- signature='s')
- if self.solicit_uuids is not None:
- properties['SolicitUUIDs'] = dbus.Array(self.solicit_uuids,
- signature='s')
- if self.manufacturer_data is not None:
- properties['ManufacturerData'] = dbus.Dictionary(
- self.manufacturer_data, signature='qay')
- if self.service_data is not None:
- properties['ServiceData'] = dbus.Dictionary(self.service_data,
- signature='say')
- return {LE_ADVERTISEMENT_IFACE: properties}
-
- def get_path(self):
- return dbus.ObjectPath(self.path)
-
- def add_service_uuid(self, uuid):
- if not self.service_uuids:
- self.service_uuids = []
- self.service_uuids.append(uuid)
-
- def add_solicit_uuid(self, uuid):
- if not self.solicit_uuids:
- self.solicit_uuids = []
- self.solicit_uuids.append(uuid)
-
- def add_manufacturer_data(self, manuf_code, data):
- if not self.manufacturer_data:
- self.manufacturer_data = dict()
- self.manufacturer_data[manuf_code] = data
-
- def add_service_data(self, uuid, data):
- if not self.service_data:
- self.service_data = dict()
- self.service_data[uuid] = data
-
- @dbus.service.method(DBUS_PROP_IFACE,
- in_signature='s',
- out_signature='a{sv}')
- def GetAll(self, interface):
- print 'GetAll'
- if interface != LE_ADVERTISEMENT_IFACE:
- raise InvalidArgsException()
- print 'returning props'
- return self.get_properties()[LE_ADVERTISEMENT_IFACE]
-
- @dbus.service.method(LE_ADVERTISEMENT_IFACE,
- in_signature='',
- out_signature='')
- def Release(self):
- print '%s: Released!' % self.path
-
-class TestAdvertisement(Advertisement):
-
- def __init__(self, bus, index):
- Advertisement.__init__(self, bus, index, 'broadcast')
- self.add_service_uuid('0000180D-0000-1000-8000-00805F9B34FB')
- self.add_service_uuid('0000180F-0000-1000-8000-00805F9B34FB')
- self.add_manufacturer_data(0xffff, [0x00, 0x01, 0x02, 0x03, 0x04])
- self.add_service_data('00009999-0000-1000-8000-00805F9B34FB',
- [0x00, 0x01, 0x02, 0x03, 0x04])
-
-
-def register_ad_cb():
- print 'Advertisement registered'
-
-
-def register_ad_error_cb(error):
- print 'Failed to register advertisement: ' + str(error)
- mainloop.quit()
-
-
-def find_adapter(bus):
- remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'),
- DBUS_OM_IFACE)
- objects = remote_om.GetManagedObjects()
-
- for o, props in objects.iteritems():
- if LE_ADVERTISING_MANAGER_IFACE in props:
- return o
-
- return None
-
-
-def main():
- global mainloop
-
- dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
-
- bus = dbus.SystemBus()
-
- adapter = find_adapter(bus)
- if not adapter:
- print 'LEAdvertisingManager1 interface not found'
- return
-
- ad_manager = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, adapter),
- LE_ADVERTISING_MANAGER_IFACE)
-
- test_advertisement = TestAdvertisement(bus, 0)
-
- mainloop = gobject.MainLoop()
-
- ad_manager.RegisterAdvertisement(test_advertisement.get_path(), {},
- reply_handler=register_ad_cb,
- error_handler=register_ad_error_cb)
-
- mainloop.run()
-
-if __name__ == '__main__':
- main()
diff --git a/test/example-advertisement b/test/example-advertisement
new file mode 100755
index 0000000..98aeafa
--- /dev/null
+++ b/test/example-advertisement
@@ -0,0 +1,170 @@
+#!/usr/bin/python
+
+import dbus
+import dbus.exceptions
+import dbus.mainloop.glib
+import dbus.service
+
+import array
+import gobject
+
+from random import randint
+
+mainloop = None
+
+BLUEZ_SERVICE_NAME = 'org.bluez'
+LE_ADVERTISING_MANAGER_IFACE = 'org.bluez.LEAdvertisingManager1'
+DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
+DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
+
+LE_ADVERTISEMENT_IFACE = 'org.bluez.LEAdvertisement1'
+
+
+class InvalidArgsException(dbus.exceptions.DBusException):
+ _dbus_error_name = 'org.freedesktop.DBus.Error.InvalidArgs'
+
+
+class NotSupportedException(dbus.exceptions.DBusException):
+ _dbus_error_name = 'org.bluez.Error.NotSupported'
+
+
+class NotPermittedException(dbus.exceptions.DBusException):
+ _dbus_error_name = 'org.bluez.Error.NotPermitted'
+
+
+class InvalidValueLengthException(dbus.exceptions.DBusException):
+ _dbus_error_name = 'org.bluez.Error.InvalidValueLength'
+
+
+class FailedException(dbus.exceptions.DBusException):
+ _dbus_error_name = 'org.bluez.Error.Failed'
+
+
+class Advertisement(dbus.service.Object):
+ PATH_BASE = '/org/bluez/example/advertisement'
+
+ def __init__(self, bus, index, advertising_type):
+ self.path = self.PATH_BASE + str(index)
+ self.bus = bus
+ self.ad_type = advertising_type
+ self.service_uuids = None
+ self.manufacturer_data = None
+ self.solicit_uuids = None
+ self.service_data = None
+ dbus.service.Object.__init__(self, bus, self.path)
+
+ def get_properties(self):
+ properties = dict()
+ properties['Type'] = self.ad_type
+ if self.service_uuids is not None:
+ properties['ServiceUUIDs'] = dbus.Array(self.service_uuids,
+ signature='s')
+ if self.solicit_uuids is not None:
+ properties['SolicitUUIDs'] = dbus.Array(self.solicit_uuids,
+ signature='s')
+ if self.manufacturer_data is not None:
+ properties['ManufacturerData'] = dbus.Dictionary(
+ self.manufacturer_data, signature='qay')
+ if self.service_data is not None:
+ properties['ServiceData'] = dbus.Dictionary(self.service_data,
+ signature='say')
+ return {LE_ADVERTISEMENT_IFACE: properties}
+
+ def get_path(self):
+ return dbus.ObjectPath(self.path)
+
+ def add_service_uuid(self, uuid):
+ if not self.service_uuids:
+ self.service_uuids = []
+ self.service_uuids.append(uuid)
+
+ def add_solicit_uuid(self, uuid):
+ if not self.solicit_uuids:
+ self.solicit_uuids = []
+ self.solicit_uuids.append(uuid)
+
+ def add_manufacturer_data(self, manuf_code, data):
+ if not self.manufacturer_data:
+ self.manufacturer_data = dict()
+ self.manufacturer_data[manuf_code] = data
+
+ def add_service_data(self, uuid, data):
+ if not self.service_data:
+ self.service_data = dict()
+ self.service_data[uuid] = data
+
+ @dbus.service.method(DBUS_PROP_IFACE,
+ in_signature='s',
+ out_signature='a{sv}')
+ def GetAll(self, interface):
+ print 'GetAll'
+ if interface != LE_ADVERTISEMENT_IFACE:
+ raise InvalidArgsException()
+ print 'returning props'
+ return self.get_properties()[LE_ADVERTISEMENT_IFACE]
+
+ @dbus.service.method(LE_ADVERTISEMENT_IFACE,
+ in_signature='',
+ out_signature='')
+ def Release(self):
+ print '%s: Released!' % self.path
+
+class TestAdvertisement(Advertisement):
+
+ def __init__(self, bus, index):
+ Advertisement.__init__(self, bus, index, 'broadcast')
+ self.add_service_uuid('0000180D-0000-1000-8000-00805F9B34FB')
+ self.add_service_uuid('0000180F-0000-1000-8000-00805F9B34FB')
+ self.add_manufacturer_data(0xffff, [0x00, 0x01, 0x02, 0x03, 0x04])
+ self.add_service_data('00009999-0000-1000-8000-00805F9B34FB',
+ [0x00, 0x01, 0x02, 0x03, 0x04])
+
+
+def register_ad_cb():
+ print 'Advertisement registered'
+
+
+def register_ad_error_cb(error):
+ print 'Failed to register advertisement: ' + str(error)
+ mainloop.quit()
+
+
+def find_adapter(bus):
+ remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'),
+ DBUS_OM_IFACE)
+ objects = remote_om.GetManagedObjects()
+
+ for o, props in objects.iteritems():
+ if LE_ADVERTISING_MANAGER_IFACE in props:
+ return o
+
+ return None
+
+
+def main():
+ global mainloop
+
+ dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
+
+ bus = dbus.SystemBus()
+
+ adapter = find_adapter(bus)
+ if not adapter:
+ print 'LEAdvertisingManager1 interface not found'
+ return
+
+ ad_manager = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, adapter),
+ LE_ADVERTISING_MANAGER_IFACE)
+
+ test_advertisement = TestAdvertisement(bus, 0)
+
+ mainloop = gobject.MainLoop()
+
+ ad_manager.RegisterAdvertisement(test_advertisement.get_path(), {},
+ reply_handler=register_ad_cb,
+ error_handler=register_ad_error_cb)
+
+ mainloop.run()
+
+if __name__ == '__main__':
+ main()
diff --git a/test/example-gatt-server b/test/example-gatt-server
new file mode 100755
index 0000000..a6f5cbe
--- /dev/null
+++ b/test/example-gatt-server
@@ -0,0 +1,533 @@
+#!/usr/bin/python
+
+import dbus
+import dbus.exceptions
+import dbus.mainloop.glib
+import dbus.service
+
+import array
+import gobject
+
+from random import randint
+
+mainloop = None
+
+BLUEZ_SERVICE_NAME = 'org.bluez'
+GATT_MANAGER_IFACE = 'org.bluez.GattManager1'
+DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
+DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
+
+GATT_SERVICE_IFACE = 'org.bluez.GattService1'
+GATT_CHRC_IFACE = 'org.bluez.GattCharacteristic1'
+GATT_DESC_IFACE = 'org.bluez.GattDescriptor1'
+
+class InvalidArgsException(dbus.exceptions.DBusException):
+ _dbus_error_name = 'org.freedesktop.DBus.Error.InvalidArgs'
+
+class NotSupportedException(dbus.exceptions.DBusException):
+ _dbus_error_name = 'org.bluez.Error.NotSupported'
+
+class NotPermittedException(dbus.exceptions.DBusException):
+ _dbus_error_name = 'org.bluez.Error.NotPermitted'
+
+class InvalidValueLengthException(dbus.exceptions.DBusException):
+ _dbus_error_name = 'org.bluez.Error.InvalidValueLength'
+
+class FailedException(dbus.exceptions.DBusException):
+ _dbus_error_name = 'org.bluez.Error.Failed'
+
+
+class Service(dbus.service.Object):
+ PATH_BASE = '/org/bluez/example/service'
+
+ def __init__(self, bus, index, uuid, primary):
+ self.path = self.PATH_BASE + str(index)
+ self.bus = bus
+ self.uuid = uuid
+ self.primary = primary
+ self.characteristics = []
+ dbus.service.Object.__init__(self, bus, self.path)
+
+ def get_properties(self):
+ return {
+ GATT_SERVICE_IFACE: {
+ 'UUID': self.uuid,
+ 'Primary': self.primary,
+ 'Characteristics': dbus.Array(
+ self.get_characteristic_paths(),
+ signature='o')
+ }
+ }
+
+ def get_path(self):
+ return dbus.ObjectPath(self.path)
+
+ def add_characteristic(self, characteristic):
+ self.characteristics.append(characteristic)
+
+ def get_characteristic_paths(self):
+ result = []
+ for chrc in self.characteristics:
+ result.append(chrc.get_path())
+ return result
+
+ def get_characteristics(self):
+ return self.characteristics
+
+ @dbus.service.method(DBUS_PROP_IFACE,
+ in_signature='s',
+ out_signature='a{sv}')
+ def GetAll(self, interface):
+ if interface != GATT_SERVICE_IFACE:
+ raise InvalidArgsException()
+
+ return self.get_properties[GATT_SERVICE_IFACE]
+
+ @dbus.service.method(DBUS_OM_IFACE, out_signature='a{oa{sa{sv}}}')
+ def GetManagedObjects(self):
+ response = {}
+ print 'GetManagedObjects'
+
+ response[self.get_path()] = self.get_properties()
+ chrcs = self.get_characteristics()
+ for chrc in chrcs:
+ response[chrc.get_path()] = chrc.get_properties()
+ descs = chrc.get_descriptors()
+ for desc in descs:
+ response[desc.get_path()] = desc.get_properties()
+
+ return response
+
+
+class Characteristic(dbus.service.Object):
+ def __init__(self, bus, index, uuid, flags, service):
+ self.path = service.path + '/char' + str(index)
+ self.bus = bus
+ self.uuid = uuid
+ self.service = service
+ self.flags = flags
+ self.descriptors = []
+ dbus.service.Object.__init__(self, bus, self.path)
+
+ def get_properties(self):
+ return {
+ GATT_CHRC_IFACE: {
+ 'Service': self.service.get_path(),
+ 'UUID': self.uuid,
+ 'Flags': self.flags,
+ 'Descriptors': dbus.Array(
+ self.get_descriptor_paths(),
+ signature='o')
+ }
+ }
+
+ def get_path(self):
+ return dbus.ObjectPath(self.path)
+
+ def add_descriptor(self, descriptor):
+ self.descriptors.append(descriptor)
+
+ def get_descriptor_paths(self):
+ result = []
+ for desc in self.descriptors:
+ result.append(desc.get_path())
+ return result
+
+ def get_descriptors(self):
+ return self.descriptors
+
+ @dbus.service.method(DBUS_PROP_IFACE,
+ in_signature='s',
+ out_signature='a{sv}')
+ def GetAll(self, interface):
+ if interface != GATT_CHRC_IFACE:
+ raise InvalidArgsException()
+
+ return self.get_properties[GATT_CHRC_IFACE]
+
+ @dbus.service.method(GATT_CHRC_IFACE, out_signature='ay')
+ def ReadValue(self):
+ print 'Default ReadValue called, returning error'
+ raise NotSupportedException()
+
+ @dbus.service.method(GATT_CHRC_IFACE, in_signature='ay')
+ def WriteValue(self, value):
+ print 'Default WriteValue called, returning error'
+ raise NotSupportedException()
+
+ @dbus.service.method(GATT_CHRC_IFACE)
+ def StartNotify(self):
+ print 'Default StartNotify called, returning error'
+ raise NotSupportedException()
+
+ @dbus.service.method(GATT_CHRC_IFACE)
+ def StopNotify(self):
+ print 'Default StopNotify called, returning error'
+ raise NotSupportedException()
+
+ @dbus.service.signal(DBUS_PROP_IFACE,
+ signature='sa{sv}as')
+ def PropertiesChanged(self, interface, changed, invalidated):
+ pass
+
+
+class Descriptor(dbus.service.Object):
+ def __init__(self, bus, index, uuid, characteristic):
+ self.path = characteristic.path + '/desc' + str(index)
+ self.bus = bus
+ self.uuid = uuid
+ self.chrc = characteristic
+ dbus.service.Object.__init__(self, bus, self.path)
+
+ def get_properties(self):
+ return {
+ GATT_DESC_IFACE: {
+ 'Characteristic': self.chrc.get_path(),
+ 'UUID': self.uuid,
+ }
+ }
+
+ def get_path(self):
+ return dbus.ObjectPath(self.path)
+
+ @dbus.service.method(DBUS_PROP_IFACE,
+ in_signature='s',
+ out_signature='a{sv}')
+ def GetAll(self, interface):
+ if interface != GATT_DESC_IFACE:
+ raise InvalidArgsException()
+
+ return self.get_properties[GATT_CHRC_IFACE]
+
+ @dbus.service.method(GATT_DESC_IFACE, out_signature='ay')
+ def ReadValue(self):
+ print 'Default ReadValue called, returning error'
+ raise NotSupportedException()
+
+ @dbus.service.method(GATT_DESC_IFACE, in_signature='ay')
+ def WriteValue(self, value):
+ print 'Default WriteValue called, returning error'
+ raise NotSupportedException()
+
+
+class HeartRateService(Service):
+ """
+ Fake Heart Rate Service that simulates a fake heart beat and control point
+ behavior.
+
+ """
+ HR_UUID = '0000180d-0000-1000-8000-00805f9b34fb'
+
+ def __init__(self, bus, index):
+ Service.__init__(self, bus, index, self.HR_UUID, True)
+ self.add_characteristic(HeartRateMeasurementChrc(bus, 0, self))
+ self.add_characteristic(BodySensorLocationChrc(bus, 1, self))
+ self.add_characteristic(HeartRateControlPointChrc(bus, 2, self))
+ self.energy_expended = 0
+
+
+class HeartRateMeasurementChrc(Characteristic):
+ HR_MSRMT_UUID = '00002a37-0000-1000-8000-00805f9b34fb'
+
+ def __init__(self, bus, index, service):
+ Characteristic.__init__(
+ self, bus, index,
+ self.HR_MSRMT_UUID,
+ ['notify'],
+ service)
+ self.notifying = False
+ self.hr_ee_count = 0
+
+ def hr_msrmt_cb(self):
+ value = []
+ value.append(dbus.Byte(0x06))
+
+ value.append(dbus.Byte(randint(90, 130)))
+
+ if self.hr_ee_count % 10 == 0:
+ value[0] = dbus.Byte(value[0] | 0x08)
+ value.append(dbus.Byte(self.service.energy_expended & 0xff))
+ value.append(dbus.Byte((self.service.energy_expended >> 8) & 0xff))
+
+ self.service.energy_expended = \
+ min(0xffff, self.service.energy_expended + 1)
+ self.hr_ee_count += 1
+
+ print 'Updating value: ' + repr(value)
+
+ self.PropertiesChanged(GATT_CHRC_IFACE, { 'Value': value }, [])
+
+ return self.notifying
+
+ def _update_hr_msrmt_simulation(self):
+ print 'Update HR Measurement Simulation'
+
+ if not self.notifying:
+ return
+
+ gobject.timeout_add(1000, self.hr_msrmt_cb)
+
+ def StartNotify(self):
+ if self.notifying:
+ print 'Already notifying, nothing to do'
+ return
+
+ self.notifying = True
+ self._update_hr_msrmt_simulation()
+
+ def StopNotify(self):
+ if not self.notifying:
+ print 'Not notifying, nothing to do'
+ return
+
+ self.notifying = False
+ self._update_hr_msrmt_simulation()
+
+
+class BodySensorLocationChrc(Characteristic):
+ BODY_SNSR_LOC_UUID = '00002a38-0000-1000-8000-00805f9b34fb'
+
+ def __init__(self, bus, index, service):
+ Characteristic.__init__(
+ self, bus, index,
+ self.BODY_SNSR_LOC_UUID,
+ ['read'],
+ service)
+
+ def ReadValue(self):
+ # Return 'Chest' as the sensor location.
+ return [ 0x01 ]
+
+class HeartRateControlPointChrc(Characteristic):
+ HR_CTRL_PT_UUID = '00002a39-0000-1000-8000-00805f9b34fb'
+
+ def __init__(self, bus, index, service):
+ Characteristic.__init__(
+ self, bus, index,
+ self.HR_CTRL_PT_UUID,
+ ['write'],
+ service)
+
+ def WriteValue(self, value):
+ print 'Heart Rate Control Point WriteValue called'
+
+ if len(value) != 1:
+ raise InvalidValueLengthException()
+
+ byte = value[0]
+ print 'Control Point value: ' + repr(byte)
+
+ if byte != 1:
+ raise FailedException("0x80")
+
+ print 'Energy Expended field reset!'
+ self.service.energy_expended = 0
+
+
+class BatteryService(Service):
+ """
+ Fake Battery service that emulates a draining battery.
+
+ """
+ BATTERY_UUID = '180f'
+
+ def __init__(self, bus, index):
+ Service.__init__(self, bus, index, self.BATTERY_UUID, True)
+ self.add_characteristic(BatteryLevelCharacteristic(bus, 0, self))
+
+
+class BatteryLevelCharacteristic(Characteristic):
+ """
+ Fake Battery Level characteristic. The battery level is drained by 2 points
+ every 5 seconds.
+
+ """
+ BATTERY_LVL_UUID = '2a19'
+
+ def __init__(self, bus, index, service):
+ Characteristic.__init__(
+ self, bus, index,
+ self.BATTERY_LVL_UUID,
+ ['read', 'notify'],
+ service)
+ self.notifying = False
+ self.battery_lvl = 100
+ gobject.timeout_add(5000, self.drain_battery)
+
+ def notify_battery_level(self):
+ if not self.notifying:
+ return
+ self.PropertiesChanged(
+ GATT_CHRC_IFACE,
+ { 'Value': [dbus.Byte(self.battery_lvl)] }, [])
+
+ def drain_battery(self):
+ if self.battery_lvl > 0:
+ self.battery_lvl -= 2
+ if self.battery_lvl < 0:
+ self.battery_lvl = 0
+ print 'Battery Level drained: ' + repr(self.battery_lvl)
+ self.notify_battery_level()
+ return True
+
+ def ReadValue(self):
+ print 'Battery Level read: ' + repr(self.battery_lvl)
+ return [dbus.Byte(self.battery_lvl)]
+
+ def StartNotify(self):
+ if self.notifying:
+ print 'Already notifying, nothing to do'
+ return
+
+ self.notifying = True
+ self.notify_battery_level()
+
+ def StopNotify(self):
+ if not self.notifying:
+ print 'Not notifying, nothing to do'
+ return
+
+ self.notifying = False
+
+
+class TestService(Service):
+ """
+ Dummy test service that provides characteristics and descriptors that
+ exercise various API functionality.
+
+ """
+ TEST_SVC_UUID = '12345678-1234-5678-1234-56789abcdef0'
+
+ def __init__(self, bus, index):
+ Service.__init__(self, bus, index, self.TEST_SVC_UUID, False)
+ self.add_characteristic(TestCharacteristic(bus, 0, self))
+
+
+class TestCharacteristic(Characteristic):
+ """
+ Dummy test characteristic. Allows writing arbitrary bytes to its value, and
+ contains "extended properties", as well as a test descriptor.
+
+ """
+ TEST_CHRC_UUID = '12345678-1234-5678-1234-56789abcdef1'
+
+ def __init__(self, bus, index, service):
+ Characteristic.__init__(
+ self, bus, index,
+ self.TEST_CHRC_UUID,
+ ['read', 'write', 'writable-auxiliaries'],
+ service)
+ self.value = []
+ self.add_descriptor(TestDescriptor(bus, 0, self))
+ self.add_descriptor(
+ CharacteristicUserDescriptionDescriptor(bus, 1, self))
+
+ def ReadValue(self):
+ print 'TestCharacteristic Read: ' + repr(self.value)
+ return self.value
+
+ def WriteValue(self, value):
+ print 'TestCharacteristic Write: ' + repr(value)
+ self.value = value
+
+
+class TestDescriptor(Descriptor):
+ """
+ Dummy test descriptor. Returns a static value.
+
+ """
+ TEST_DESC_UUID = '12345678-1234-5678-1234-56789abcdef2'
+
+ def __init__(self, bus, index, characteristic):
+ Descriptor.__init__(
+ self, bus, index,
+ self.TEST_DESC_UUID,
+ characteristic)
+
+ def ReadValue(self):
+ return [
+ dbus.Byte('T'), dbus.Byte('e'), dbus.Byte('s'), dbus.Byte('t')
+ ]
+
+
+class CharacteristicUserDescriptionDescriptor(Descriptor):
+ """
+ Writable CUD descriptor.
+
+ """
+ CUD_UUID = '2901'
+
+ def __init__(self, bus, index, characteristic):
+ self.writable = 'writable-auxiliaries' in characteristic.flags
+ self.value = array.array('B', 'This is a characteristic for testing')
+ self.value = self.value.tolist()
+ Descriptor.__init__(
+ self, bus, index,
+ self.CUD_UUID,
+ characteristic)
+
+ def ReadValue(self):
+ return self.value
+
+ def WriteValue(self, value):
+ if not self.writable:
+ raise NotPermittedException()
+ self.value = value
+
+
+def register_service_cb():
+ print 'GATT service registered'
+
+
+def register_service_error_cb(error):
+ print 'Failed to register service: ' + str(error)
+ mainloop.quit()
+
+
+def find_adapter(bus):
+ remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'),
+ DBUS_OM_IFACE)
+ objects = remote_om.GetManagedObjects()
+
+ for o, props in objects.iteritems():
+ if props.has_key(GATT_MANAGER_IFACE):
+ return o
+
+ return None
+
+def main():
+ global mainloop
+
+ dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
+
+ bus = dbus.SystemBus()
+
+ adapter = find_adapter(bus)
+ if not adapter:
+ print 'GattManager1 interface not found'
+ return
+
+ service_manager = dbus.Interface(
+ bus.get_object(BLUEZ_SERVICE_NAME, adapter),
+ GATT_MANAGER_IFACE)
+
+ hr_service = HeartRateService(bus, 0)
+ bat_service = BatteryService(bus, 1)
+ test_service = TestService(bus, 2)
+
+ mainloop = gobject.MainLoop()
+
+ service_manager.RegisterService(hr_service.get_path(), {},
+ reply_handler=register_service_cb,
+ error_handler=register_service_error_cb)
+ service_manager.RegisterService(bat_service.get_path(), {},
+ reply_handler=register_service_cb,
+ error_handler=register_service_error_cb)
+ service_manager.RegisterService(test_service.get_path(), {},
+ reply_handler=register_service_cb,
+ error_handler=register_service_error_cb)
+
+ mainloop.run()
+
+if __name__ == '__main__':
+ main()
diff --git a/tools/gatt-example b/tools/gatt-example
deleted file mode 100644
index a6f5cbe..0000000
--- a/tools/gatt-example
+++ /dev/null
@@ -1,533 +0,0 @@
-#!/usr/bin/python
-
-import dbus
-import dbus.exceptions
-import dbus.mainloop.glib
-import dbus.service
-
-import array
-import gobject
-
-from random import randint
-
-mainloop = None
-
-BLUEZ_SERVICE_NAME = 'org.bluez'
-GATT_MANAGER_IFACE = 'org.bluez.GattManager1'
-DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
-DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
-
-GATT_SERVICE_IFACE = 'org.bluez.GattService1'
-GATT_CHRC_IFACE = 'org.bluez.GattCharacteristic1'
-GATT_DESC_IFACE = 'org.bluez.GattDescriptor1'
-
-class InvalidArgsException(dbus.exceptions.DBusException):
- _dbus_error_name = 'org.freedesktop.DBus.Error.InvalidArgs'
-
-class NotSupportedException(dbus.exceptions.DBusException):
- _dbus_error_name = 'org.bluez.Error.NotSupported'
-
-class NotPermittedException(dbus.exceptions.DBusException):
- _dbus_error_name = 'org.bluez.Error.NotPermitted'
-
-class InvalidValueLengthException(dbus.exceptions.DBusException):
- _dbus_error_name = 'org.bluez.Error.InvalidValueLength'
-
-class FailedException(dbus.exceptions.DBusException):
- _dbus_error_name = 'org.bluez.Error.Failed'
-
-
-class Service(dbus.service.Object):
- PATH_BASE = '/org/bluez/example/service'
-
- def __init__(self, bus, index, uuid, primary):
- self.path = self.PATH_BASE + str(index)
- self.bus = bus
- self.uuid = uuid
- self.primary = primary
- self.characteristics = []
- dbus.service.Object.__init__(self, bus, self.path)
-
- def get_properties(self):
- return {
- GATT_SERVICE_IFACE: {
- 'UUID': self.uuid,
- 'Primary': self.primary,
- 'Characteristics': dbus.Array(
- self.get_characteristic_paths(),
- signature='o')
- }
- }
-
- def get_path(self):
- return dbus.ObjectPath(self.path)
-
- def add_characteristic(self, characteristic):
- self.characteristics.append(characteristic)
-
- def get_characteristic_paths(self):
- result = []
- for chrc in self.characteristics:
- result.append(chrc.get_path())
- return result
-
- def get_characteristics(self):
- return self.characteristics
-
- @dbus.service.method(DBUS_PROP_IFACE,
- in_signature='s',
- out_signature='a{sv}')
- def GetAll(self, interface):
- if interface != GATT_SERVICE_IFACE:
- raise InvalidArgsException()
-
- return self.get_properties[GATT_SERVICE_IFACE]
-
- @dbus.service.method(DBUS_OM_IFACE, out_signature='a{oa{sa{sv}}}')
- def GetManagedObjects(self):
- response = {}
- print 'GetManagedObjects'
-
- response[self.get_path()] = self.get_properties()
- chrcs = self.get_characteristics()
- for chrc in chrcs:
- response[chrc.get_path()] = chrc.get_properties()
- descs = chrc.get_descriptors()
- for desc in descs:
- response[desc.get_path()] = desc.get_properties()
-
- return response
-
-
-class Characteristic(dbus.service.Object):
- def __init__(self, bus, index, uuid, flags, service):
- self.path = service.path + '/char' + str(index)
- self.bus = bus
- self.uuid = uuid
- self.service = service
- self.flags = flags
- self.descriptors = []
- dbus.service.Object.__init__(self, bus, self.path)
-
- def get_properties(self):
- return {
- GATT_CHRC_IFACE: {
- 'Service': self.service.get_path(),
- 'UUID': self.uuid,
- 'Flags': self.flags,
- 'Descriptors': dbus.Array(
- self.get_descriptor_paths(),
- signature='o')
- }
- }
-
- def get_path(self):
- return dbus.ObjectPath(self.path)
-
- def add_descriptor(self, descriptor):
- self.descriptors.append(descriptor)
-
- def get_descriptor_paths(self):
- result = []
- for desc in self.descriptors:
- result.append(desc.get_path())
- return result
-
- def get_descriptors(self):
- return self.descriptors
-
- @dbus.service.method(DBUS_PROP_IFACE,
- in_signature='s',
- out_signature='a{sv}')
- def GetAll(self, interface):
- if interface != GATT_CHRC_IFACE:
- raise InvalidArgsException()
-
- return self.get_properties[GATT_CHRC_IFACE]
-
- @dbus.service.method(GATT_CHRC_IFACE, out_signature='ay')
- def ReadValue(self):
- print 'Default ReadValue called, returning error'
- raise NotSupportedException()
-
- @dbus.service.method(GATT_CHRC_IFACE, in_signature='ay')
- def WriteValue(self, value):
- print 'Default WriteValue called, returning error'
- raise NotSupportedException()
-
- @dbus.service.method(GATT_CHRC_IFACE)
- def StartNotify(self):
- print 'Default StartNotify called, returning error'
- raise NotSupportedException()
-
- @dbus.service.method(GATT_CHRC_IFACE)
- def StopNotify(self):
- print 'Default StopNotify called, returning error'
- raise NotSupportedException()
-
- @dbus.service.signal(DBUS_PROP_IFACE,
- signature='sa{sv}as')
- def PropertiesChanged(self, interface, changed, invalidated):
- pass
-
-
-class Descriptor(dbus.service.Object):
- def __init__(self, bus, index, uuid, characteristic):
- self.path = characteristic.path + '/desc' + str(index)
- self.bus = bus
- self.uuid = uuid
- self.chrc = characteristic
- dbus.service.Object.__init__(self, bus, self.path)
-
- def get_properties(self):
- return {
- GATT_DESC_IFACE: {
- 'Characteristic': self.chrc.get_path(),
- 'UUID': self.uuid,
- }
- }
-
- def get_path(self):
- return dbus.ObjectPath(self.path)
-
- @dbus.service.method(DBUS_PROP_IFACE,
- in_signature='s',
- out_signature='a{sv}')
- def GetAll(self, interface):
- if interface != GATT_DESC_IFACE:
- raise InvalidArgsException()
-
- return self.get_properties[GATT_CHRC_IFACE]
-
- @dbus.service.method(GATT_DESC_IFACE, out_signature='ay')
- def ReadValue(self):
- print 'Default ReadValue called, returning error'
- raise NotSupportedException()
-
- @dbus.service.method(GATT_DESC_IFACE, in_signature='ay')
- def WriteValue(self, value):
- print 'Default WriteValue called, returning error'
- raise NotSupportedException()
-
-
-class HeartRateService(Service):
- """
- Fake Heart Rate Service that simulates a fake heart beat and control point
- behavior.
-
- """
- HR_UUID = '0000180d-0000-1000-8000-00805f9b34fb'
-
- def __init__(self, bus, index):
- Service.__init__(self, bus, index, self.HR_UUID, True)
- self.add_characteristic(HeartRateMeasurementChrc(bus, 0, self))
- self.add_characteristic(BodySensorLocationChrc(bus, 1, self))
- self.add_characteristic(HeartRateControlPointChrc(bus, 2, self))
- self.energy_expended = 0
-
-
-class HeartRateMeasurementChrc(Characteristic):
- HR_MSRMT_UUID = '00002a37-0000-1000-8000-00805f9b34fb'
-
- def __init__(self, bus, index, service):
- Characteristic.__init__(
- self, bus, index,
- self.HR_MSRMT_UUID,
- ['notify'],
- service)
- self.notifying = False
- self.hr_ee_count = 0
-
- def hr_msrmt_cb(self):
- value = []
- value.append(dbus.Byte(0x06))
-
- value.append(dbus.Byte(randint(90, 130)))
-
- if self.hr_ee_count % 10 == 0:
- value[0] = dbus.Byte(value[0] | 0x08)
- value.append(dbus.Byte(self.service.energy_expended & 0xff))
- value.append(dbus.Byte((self.service.energy_expended >> 8) & 0xff))
-
- self.service.energy_expended = \
- min(0xffff, self.service.energy_expended + 1)
- self.hr_ee_count += 1
-
- print 'Updating value: ' + repr(value)
-
- self.PropertiesChanged(GATT_CHRC_IFACE, { 'Value': value }, [])
-
- return self.notifying
-
- def _update_hr_msrmt_simulation(self):
- print 'Update HR Measurement Simulation'
-
- if not self.notifying:
- return
-
- gobject.timeout_add(1000, self.hr_msrmt_cb)
-
- def StartNotify(self):
- if self.notifying:
- print 'Already notifying, nothing to do'
- return
-
- self.notifying = True
- self._update_hr_msrmt_simulation()
-
- def StopNotify(self):
- if not self.notifying:
- print 'Not notifying, nothing to do'
- return
-
- self.notifying = False
- self._update_hr_msrmt_simulation()
-
-
-class BodySensorLocationChrc(Characteristic):
- BODY_SNSR_LOC_UUID = '00002a38-0000-1000-8000-00805f9b34fb'
-
- def __init__(self, bus, index, service):
- Characteristic.__init__(
- self, bus, index,
- self.BODY_SNSR_LOC_UUID,
- ['read'],
- service)
-
- def ReadValue(self):
- # Return 'Chest' as the sensor location.
- return [ 0x01 ]
-
-class HeartRateControlPointChrc(Characteristic):
- HR_CTRL_PT_UUID = '00002a39-0000-1000-8000-00805f9b34fb'
-
- def __init__(self, bus, index, service):
- Characteristic.__init__(
- self, bus, index,
- self.HR_CTRL_PT_UUID,
- ['write'],
- service)
-
- def WriteValue(self, value):
- print 'Heart Rate Control Point WriteValue called'
-
- if len(value) != 1:
- raise InvalidValueLengthException()
-
- byte = value[0]
- print 'Control Point value: ' + repr(byte)
-
- if byte != 1:
- raise FailedException("0x80")
-
- print 'Energy Expended field reset!'
- self.service.energy_expended = 0
-
-
-class BatteryService(Service):
- """
- Fake Battery service that emulates a draining battery.
-
- """
- BATTERY_UUID = '180f'
-
- def __init__(self, bus, index):
- Service.__init__(self, bus, index, self.BATTERY_UUID, True)
- self.add_characteristic(BatteryLevelCharacteristic(bus, 0, self))
-
-
-class BatteryLevelCharacteristic(Characteristic):
- """
- Fake Battery Level characteristic. The battery level is drained by 2 points
- every 5 seconds.
-
- """
- BATTERY_LVL_UUID = '2a19'
-
- def __init__(self, bus, index, service):
- Characteristic.__init__(
- self, bus, index,
- self.BATTERY_LVL_UUID,
- ['read', 'notify'],
- service)
- self.notifying = False
- self.battery_lvl = 100
- gobject.timeout_add(5000, self.drain_battery)
-
- def notify_battery_level(self):
- if not self.notifying:
- return
- self.PropertiesChanged(
- GATT_CHRC_IFACE,
- { 'Value': [dbus.Byte(self.battery_lvl)] }, [])
-
- def drain_battery(self):
- if self.battery_lvl > 0:
- self.battery_lvl -= 2
- if self.battery_lvl < 0:
- self.battery_lvl = 0
- print 'Battery Level drained: ' + repr(self.battery_lvl)
- self.notify_battery_level()
- return True
-
- def ReadValue(self):
- print 'Battery Level read: ' + repr(self.battery_lvl)
- return [dbus.Byte(self.battery_lvl)]
-
- def StartNotify(self):
- if self.notifying:
- print 'Already notifying, nothing to do'
- return
-
- self.notifying = True
- self.notify_battery_level()
-
- def StopNotify(self):
- if not self.notifying:
- print 'Not notifying, nothing to do'
- return
-
- self.notifying = False
-
-
-class TestService(Service):
- """
- Dummy test service that provides characteristics and descriptors that
- exercise various API functionality.
-
- """
- TEST_SVC_UUID = '12345678-1234-5678-1234-56789abcdef0'
-
- def __init__(self, bus, index):
- Service.__init__(self, bus, index, self.TEST_SVC_UUID, False)
- self.add_characteristic(TestCharacteristic(bus, 0, self))
-
-
-class TestCharacteristic(Characteristic):
- """
- Dummy test characteristic. Allows writing arbitrary bytes to its value, and
- contains "extended properties", as well as a test descriptor.
-
- """
- TEST_CHRC_UUID = '12345678-1234-5678-1234-56789abcdef1'
-
- def __init__(self, bus, index, service):
- Characteristic.__init__(
- self, bus, index,
- self.TEST_CHRC_UUID,
- ['read', 'write', 'writable-auxiliaries'],
- service)
- self.value = []
- self.add_descriptor(TestDescriptor(bus, 0, self))
- self.add_descriptor(
- CharacteristicUserDescriptionDescriptor(bus, 1, self))
-
- def ReadValue(self):
- print 'TestCharacteristic Read: ' + repr(self.value)
- return self.value
-
- def WriteValue(self, value):
- print 'TestCharacteristic Write: ' + repr(value)
- self.value = value
-
-
-class TestDescriptor(Descriptor):
- """
- Dummy test descriptor. Returns a static value.
-
- """
- TEST_DESC_UUID = '12345678-1234-5678-1234-56789abcdef2'
-
- def __init__(self, bus, index, characteristic):
- Descriptor.__init__(
- self, bus, index,
- self.TEST_DESC_UUID,
- characteristic)
-
- def ReadValue(self):
- return [
- dbus.Byte('T'), dbus.Byte('e'), dbus.Byte('s'), dbus.Byte('t')
- ]
-
-
-class CharacteristicUserDescriptionDescriptor(Descriptor):
- """
- Writable CUD descriptor.
-
- """
- CUD_UUID = '2901'
-
- def __init__(self, bus, index, characteristic):
- self.writable = 'writable-auxiliaries' in characteristic.flags
- self.value = array.array('B', 'This is a characteristic for testing')
- self.value = self.value.tolist()
- Descriptor.__init__(
- self, bus, index,
- self.CUD_UUID,
- characteristic)
-
- def ReadValue(self):
- return self.value
-
- def WriteValue(self, value):
- if not self.writable:
- raise NotPermittedException()
- self.value = value
-
-
-def register_service_cb():
- print 'GATT service registered'
-
-
-def register_service_error_cb(error):
- print 'Failed to register service: ' + str(error)
- mainloop.quit()
-
-
-def find_adapter(bus):
- remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'),
- DBUS_OM_IFACE)
- objects = remote_om.GetManagedObjects()
-
- for o, props in objects.iteritems():
- if props.has_key(GATT_MANAGER_IFACE):
- return o
-
- return None
-
-def main():
- global mainloop
-
- dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
-
- bus = dbus.SystemBus()
-
- adapter = find_adapter(bus)
- if not adapter:
- print 'GattManager1 interface not found'
- return
-
- service_manager = dbus.Interface(
- bus.get_object(BLUEZ_SERVICE_NAME, adapter),
- GATT_MANAGER_IFACE)
-
- hr_service = HeartRateService(bus, 0)
- bat_service = BatteryService(bus, 1)
- test_service = TestService(bus, 2)
-
- mainloop = gobject.MainLoop()
-
- service_manager.RegisterService(hr_service.get_path(), {},
- reply_handler=register_service_cb,
- error_handler=register_service_error_cb)
- service_manager.RegisterService(bat_service.get_path(), {},
- reply_handler=register_service_cb,
- error_handler=register_service_error_cb)
- service_manager.RegisterService(test_service.get_path(), {},
- reply_handler=register_service_cb,
- error_handler=register_service_error_cb)
-
- mainloop.run()
-
-if __name__ == '__main__':
- main()
--
2.2.0.rc0.207.ga3a616c
Hi,
> On Tue, Mar 31, 2015 at 5:53 PM, Arman Uguray <[email protected]> wrote:
> This patch moves tools/gatt-example to test/ and renames it to
> example-gatt-server in preparation for example-gatt-client.
> The same naming convention (example-*) is also adopted for
> test/advertisement-example, so it's now called
> test/example-advertisement. Also, both scripts now have the execute
> set appropriately to be consistent with the other Python scripts.
> ---
> test/advertisement-example | 170 ---------------
> test/example-advertisement | 170 +++++++++++++++
> test/example-gatt-server | 533 +++++++++++++++++++++++++++++++++++++++++++++
> tools/gatt-example | 533 ---------------------------------------------
> 4 files changed, 703 insertions(+), 703 deletions(-)
> delete mode 100644 test/advertisement-example
> create mode 100755 test/example-advertisement
> create mode 100755 test/example-gatt-server
> delete mode 100644 tools/gatt-example
>
> diff --git a/test/advertisement-example b/test/advertisement-example
> deleted file mode 100644
> index 98aeafa..0000000
> --- a/test/advertisement-example
> +++ /dev/null
> @@ -1,170 +0,0 @@
> -#!/usr/bin/python
> -
> -import dbus
> -import dbus.exceptions
> -import dbus.mainloop.glib
> -import dbus.service
> -
> -import array
> -import gobject
> -
> -from random import randint
> -
> -mainloop = None
> -
> -BLUEZ_SERVICE_NAME = 'org.bluez'
> -LE_ADVERTISING_MANAGER_IFACE = 'org.bluez.LEAdvertisingManager1'
> -DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
> -DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
> -
> -LE_ADVERTISEMENT_IFACE = 'org.bluez.LEAdvertisement1'
> -
> -
> -class InvalidArgsException(dbus.exceptions.DBusException):
> - _dbus_error_name = 'org.freedesktop.DBus.Error.InvalidArgs'
> -
> -
> -class NotSupportedException(dbus.exceptions.DBusException):
> - _dbus_error_name = 'org.bluez.Error.NotSupported'
> -
> -
> -class NotPermittedException(dbus.exceptions.DBusException):
> - _dbus_error_name = 'org.bluez.Error.NotPermitted'
> -
> -
> -class InvalidValueLengthException(dbus.exceptions.DBusException):
> - _dbus_error_name = 'org.bluez.Error.InvalidValueLength'
> -
> -
> -class FailedException(dbus.exceptions.DBusException):
> - _dbus_error_name = 'org.bluez.Error.Failed'
> -
> -
> -class Advertisement(dbus.service.Object):
> - PATH_BASE = '/org/bluez/example/advertisement'
> -
> - def __init__(self, bus, index, advertising_type):
> - self.path = self.PATH_BASE + str(index)
> - self.bus = bus
> - self.ad_type = advertising_type
> - self.service_uuids = None
> - self.manufacturer_data = None
> - self.solicit_uuids = None
> - self.service_data = None
> - dbus.service.Object.__init__(self, bus, self.path)
> -
> - def get_properties(self):
> - properties = dict()
> - properties['Type'] = self.ad_type
> - if self.service_uuids is not None:
> - properties['ServiceUUIDs'] = dbus.Array(self.service_uuids,
> - signature='s')
> - if self.solicit_uuids is not None:
> - properties['SolicitUUIDs'] = dbus.Array(self.solicit_uuids,
> - signature='s')
> - if self.manufacturer_data is not None:
> - properties['ManufacturerData'] = dbus.Dictionary(
> - self.manufacturer_data, signature='qay')
> - if self.service_data is not None:
> - properties['ServiceData'] = dbus.Dictionary(self.service_data,
> - signature='say')
> - return {LE_ADVERTISEMENT_IFACE: properties}
> -
> - def get_path(self):
> - return dbus.ObjectPath(self.path)
> -
> - def add_service_uuid(self, uuid):
> - if not self.service_uuids:
> - self.service_uuids = []
> - self.service_uuids.append(uuid)
> -
> - def add_solicit_uuid(self, uuid):
> - if not self.solicit_uuids:
> - self.solicit_uuids = []
> - self.solicit_uuids.append(uuid)
> -
> - def add_manufacturer_data(self, manuf_code, data):
> - if not self.manufacturer_data:
> - self.manufacturer_data = dict()
> - self.manufacturer_data[manuf_code] = data
> -
> - def add_service_data(self, uuid, data):
> - if not self.service_data:
> - self.service_data = dict()
> - self.service_data[uuid] = data
> -
> - @dbus.service.method(DBUS_PROP_IFACE,
> - in_signature='s',
> - out_signature='a{sv}')
> - def GetAll(self, interface):
> - print 'GetAll'
> - if interface != LE_ADVERTISEMENT_IFACE:
> - raise InvalidArgsException()
> - print 'returning props'
> - return self.get_properties()[LE_ADVERTISEMENT_IFACE]
> -
> - @dbus.service.method(LE_ADVERTISEMENT_IFACE,
> - in_signature='',
> - out_signature='')
> - def Release(self):
> - print '%s: Released!' % self.path
> -
> -class TestAdvertisement(Advertisement):
> -
> - def __init__(self, bus, index):
> - Advertisement.__init__(self, bus, index, 'broadcast')
> - self.add_service_uuid('0000180D-0000-1000-8000-00805F9B34FB')
> - self.add_service_uuid('0000180F-0000-1000-8000-00805F9B34FB')
> - self.add_manufacturer_data(0xffff, [0x00, 0x01, 0x02, 0x03, 0x04])
> - self.add_service_data('00009999-0000-1000-8000-00805F9B34FB',
> - [0x00, 0x01, 0x02, 0x03, 0x04])
> -
> -
> -def register_ad_cb():
> - print 'Advertisement registered'
> -
> -
> -def register_ad_error_cb(error):
> - print 'Failed to register advertisement: ' + str(error)
> - mainloop.quit()
> -
> -
> -def find_adapter(bus):
> - remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'),
> - DBUS_OM_IFACE)
> - objects = remote_om.GetManagedObjects()
> -
> - for o, props in objects.iteritems():
> - if LE_ADVERTISING_MANAGER_IFACE in props:
> - return o
> -
> - return None
> -
> -
> -def main():
> - global mainloop
> -
> - dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
> -
> - bus = dbus.SystemBus()
> -
> - adapter = find_adapter(bus)
> - if not adapter:
> - print 'LEAdvertisingManager1 interface not found'
> - return
> -
> - ad_manager = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, adapter),
> - LE_ADVERTISING_MANAGER_IFACE)
> -
> - test_advertisement = TestAdvertisement(bus, 0)
> -
> - mainloop = gobject.MainLoop()
> -
> - ad_manager.RegisterAdvertisement(test_advertisement.get_path(), {},
> - reply_handler=register_ad_cb,
> - error_handler=register_ad_error_cb)
> -
> - mainloop.run()
> -
> -if __name__ == '__main__':
> - main()
> diff --git a/test/example-advertisement b/test/example-advertisement
> new file mode 100755
> index 0000000..98aeafa
> --- /dev/null
> +++ b/test/example-advertisement
> @@ -0,0 +1,170 @@
> +#!/usr/bin/python
> +
> +import dbus
> +import dbus.exceptions
> +import dbus.mainloop.glib
> +import dbus.service
> +
> +import array
> +import gobject
> +
> +from random import randint
> +
> +mainloop = None
> +
> +BLUEZ_SERVICE_NAME = 'org.bluez'
> +LE_ADVERTISING_MANAGER_IFACE = 'org.bluez.LEAdvertisingManager1'
> +DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
> +DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
> +
> +LE_ADVERTISEMENT_IFACE = 'org.bluez.LEAdvertisement1'
> +
> +
> +class InvalidArgsException(dbus.exceptions.DBusException):
> + _dbus_error_name = 'org.freedesktop.DBus.Error.InvalidArgs'
> +
> +
> +class NotSupportedException(dbus.exceptions.DBusException):
> + _dbus_error_name = 'org.bluez.Error.NotSupported'
> +
> +
> +class NotPermittedException(dbus.exceptions.DBusException):
> + _dbus_error_name = 'org.bluez.Error.NotPermitted'
> +
> +
> +class InvalidValueLengthException(dbus.exceptions.DBusException):
> + _dbus_error_name = 'org.bluez.Error.InvalidValueLength'
> +
> +
> +class FailedException(dbus.exceptions.DBusException):
> + _dbus_error_name = 'org.bluez.Error.Failed'
> +
> +
> +class Advertisement(dbus.service.Object):
> + PATH_BASE = '/org/bluez/example/advertisement'
> +
> + def __init__(self, bus, index, advertising_type):
> + self.path = self.PATH_BASE + str(index)
> + self.bus = bus
> + self.ad_type = advertising_type
> + self.service_uuids = None
> + self.manufacturer_data = None
> + self.solicit_uuids = None
> + self.service_data = None
> + dbus.service.Object.__init__(self, bus, self.path)
> +
> + def get_properties(self):
> + properties = dict()
> + properties['Type'] = self.ad_type
> + if self.service_uuids is not None:
> + properties['ServiceUUIDs'] = dbus.Array(self.service_uuids,
> + signature='s')
> + if self.solicit_uuids is not None:
> + properties['SolicitUUIDs'] = dbus.Array(self.solicit_uuids,
> + signature='s')
> + if self.manufacturer_data is not None:
> + properties['ManufacturerData'] = dbus.Dictionary(
> + self.manufacturer_data, signature='qay')
> + if self.service_data is not None:
> + properties['ServiceData'] = dbus.Dictionary(self.service_data,
> + signature='say')
> + return {LE_ADVERTISEMENT_IFACE: properties}
> +
> + def get_path(self):
> + return dbus.ObjectPath(self.path)
> +
> + def add_service_uuid(self, uuid):
> + if not self.service_uuids:
> + self.service_uuids = []
> + self.service_uuids.append(uuid)
> +
> + def add_solicit_uuid(self, uuid):
> + if not self.solicit_uuids:
> + self.solicit_uuids = []
> + self.solicit_uuids.append(uuid)
> +
> + def add_manufacturer_data(self, manuf_code, data):
> + if not self.manufacturer_data:
> + self.manufacturer_data = dict()
> + self.manufacturer_data[manuf_code] = data
> +
> + def add_service_data(self, uuid, data):
> + if not self.service_data:
> + self.service_data = dict()
> + self.service_data[uuid] = data
> +
> + @dbus.service.method(DBUS_PROP_IFACE,
> + in_signature='s',
> + out_signature='a{sv}')
> + def GetAll(self, interface):
> + print 'GetAll'
> + if interface != LE_ADVERTISEMENT_IFACE:
> + raise InvalidArgsException()
> + print 'returning props'
> + return self.get_properties()[LE_ADVERTISEMENT_IFACE]
> +
> + @dbus.service.method(LE_ADVERTISEMENT_IFACE,
> + in_signature='',
> + out_signature='')
> + def Release(self):
> + print '%s: Released!' % self.path
> +
> +class TestAdvertisement(Advertisement):
> +
> + def __init__(self, bus, index):
> + Advertisement.__init__(self, bus, index, 'broadcast')
> + self.add_service_uuid('0000180D-0000-1000-8000-00805F9B34FB')
> + self.add_service_uuid('0000180F-0000-1000-8000-00805F9B34FB')
> + self.add_manufacturer_data(0xffff, [0x00, 0x01, 0x02, 0x03, 0x04])
> + self.add_service_data('00009999-0000-1000-8000-00805F9B34FB',
> + [0x00, 0x01, 0x02, 0x03, 0x04])
> +
> +
> +def register_ad_cb():
> + print 'Advertisement registered'
> +
> +
> +def register_ad_error_cb(error):
> + print 'Failed to register advertisement: ' + str(error)
> + mainloop.quit()
> +
> +
> +def find_adapter(bus):
> + remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'),
> + DBUS_OM_IFACE)
> + objects = remote_om.GetManagedObjects()
> +
> + for o, props in objects.iteritems():
> + if LE_ADVERTISING_MANAGER_IFACE in props:
> + return o
> +
> + return None
> +
> +
> +def main():
> + global mainloop
> +
> + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
> +
> + bus = dbus.SystemBus()
> +
> + adapter = find_adapter(bus)
> + if not adapter:
> + print 'LEAdvertisingManager1 interface not found'
> + return
> +
> + ad_manager = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, adapter),
> + LE_ADVERTISING_MANAGER_IFACE)
> +
> + test_advertisement = TestAdvertisement(bus, 0)
> +
> + mainloop = gobject.MainLoop()
> +
> + ad_manager.RegisterAdvertisement(test_advertisement.get_path(), {},
> + reply_handler=register_ad_cb,
> + error_handler=register_ad_error_cb)
> +
> + mainloop.run()
> +
> +if __name__ == '__main__':
> + main()
> diff --git a/test/example-gatt-server b/test/example-gatt-server
> new file mode 100755
> index 0000000..a6f5cbe
> --- /dev/null
> +++ b/test/example-gatt-server
> @@ -0,0 +1,533 @@
> +#!/usr/bin/python
> +
> +import dbus
> +import dbus.exceptions
> +import dbus.mainloop.glib
> +import dbus.service
> +
> +import array
> +import gobject
> +
> +from random import randint
> +
> +mainloop = None
> +
> +BLUEZ_SERVICE_NAME = 'org.bluez'
> +GATT_MANAGER_IFACE = 'org.bluez.GattManager1'
> +DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
> +DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
> +
> +GATT_SERVICE_IFACE = 'org.bluez.GattService1'
> +GATT_CHRC_IFACE = 'org.bluez.GattCharacteristic1'
> +GATT_DESC_IFACE = 'org.bluez.GattDescriptor1'
> +
> +class InvalidArgsException(dbus.exceptions.DBusException):
> + _dbus_error_name = 'org.freedesktop.DBus.Error.InvalidArgs'
> +
> +class NotSupportedException(dbus.exceptions.DBusException):
> + _dbus_error_name = 'org.bluez.Error.NotSupported'
> +
> +class NotPermittedException(dbus.exceptions.DBusException):
> + _dbus_error_name = 'org.bluez.Error.NotPermitted'
> +
> +class InvalidValueLengthException(dbus.exceptions.DBusException):
> + _dbus_error_name = 'org.bluez.Error.InvalidValueLength'
> +
> +class FailedException(dbus.exceptions.DBusException):
> + _dbus_error_name = 'org.bluez.Error.Failed'
> +
> +
> +class Service(dbus.service.Object):
> + PATH_BASE = '/org/bluez/example/service'
> +
> + def __init__(self, bus, index, uuid, primary):
> + self.path = self.PATH_BASE + str(index)
> + self.bus = bus
> + self.uuid = uuid
> + self.primary = primary
> + self.characteristics = []
> + dbus.service.Object.__init__(self, bus, self.path)
> +
> + def get_properties(self):
> + return {
> + GATT_SERVICE_IFACE: {
> + 'UUID': self.uuid,
> + 'Primary': self.primary,
> + 'Characteristics': dbus.Array(
> + self.get_characteristic_paths(),
> + signature='o')
> + }
> + }
> +
> + def get_path(self):
> + return dbus.ObjectPath(self.path)
> +
> + def add_characteristic(self, characteristic):
> + self.characteristics.append(characteristic)
> +
> + def get_characteristic_paths(self):
> + result = []
> + for chrc in self.characteristics:
> + result.append(chrc.get_path())
> + return result
> +
> + def get_characteristics(self):
> + return self.characteristics
> +
> + @dbus.service.method(DBUS_PROP_IFACE,
> + in_signature='s',
> + out_signature='a{sv}')
> + def GetAll(self, interface):
> + if interface != GATT_SERVICE_IFACE:
> + raise InvalidArgsException()
> +
> + return self.get_properties[GATT_SERVICE_IFACE]
> +
> + @dbus.service.method(DBUS_OM_IFACE, out_signature='a{oa{sa{sv}}}')
> + def GetManagedObjects(self):
> + response = {}
> + print 'GetManagedObjects'
> +
> + response[self.get_path()] = self.get_properties()
> + chrcs = self.get_characteristics()
> + for chrc in chrcs:
> + response[chrc.get_path()] = chrc.get_properties()
> + descs = chrc.get_descriptors()
> + for desc in descs:
> + response[desc.get_path()] = desc.get_properties()
> +
> + return response
> +
> +
> +class Characteristic(dbus.service.Object):
> + def __init__(self, bus, index, uuid, flags, service):
> + self.path = service.path + '/char' + str(index)
> + self.bus = bus
> + self.uuid = uuid
> + self.service = service
> + self.flags = flags
> + self.descriptors = []
> + dbus.service.Object.__init__(self, bus, self.path)
> +
> + def get_properties(self):
> + return {
> + GATT_CHRC_IFACE: {
> + 'Service': self.service.get_path(),
> + 'UUID': self.uuid,
> + 'Flags': self.flags,
> + 'Descriptors': dbus.Array(
> + self.get_descriptor_paths(),
> + signature='o')
> + }
> + }
> +
> + def get_path(self):
> + return dbus.ObjectPath(self.path)
> +
> + def add_descriptor(self, descriptor):
> + self.descriptors.append(descriptor)
> +
> + def get_descriptor_paths(self):
> + result = []
> + for desc in self.descriptors:
> + result.append(desc.get_path())
> + return result
> +
> + def get_descriptors(self):
> + return self.descriptors
> +
> + @dbus.service.method(DBUS_PROP_IFACE,
> + in_signature='s',
> + out_signature='a{sv}')
> + def GetAll(self, interface):
> + if interface != GATT_CHRC_IFACE:
> + raise InvalidArgsException()
> +
> + return self.get_properties[GATT_CHRC_IFACE]
> +
> + @dbus.service.method(GATT_CHRC_IFACE, out_signature='ay')
> + def ReadValue(self):
> + print 'Default ReadValue called, returning error'
> + raise NotSupportedException()
> +
> + @dbus.service.method(GATT_CHRC_IFACE, in_signature='ay')
> + def WriteValue(self, value):
> + print 'Default WriteValue called, returning error'
> + raise NotSupportedException()
> +
> + @dbus.service.method(GATT_CHRC_IFACE)
> + def StartNotify(self):
> + print 'Default StartNotify called, returning error'
> + raise NotSupportedException()
> +
> + @dbus.service.method(GATT_CHRC_IFACE)
> + def StopNotify(self):
> + print 'Default StopNotify called, returning error'
> + raise NotSupportedException()
> +
> + @dbus.service.signal(DBUS_PROP_IFACE,
> + signature='sa{sv}as')
> + def PropertiesChanged(self, interface, changed, invalidated):
> + pass
> +
> +
> +class Descriptor(dbus.service.Object):
> + def __init__(self, bus, index, uuid, characteristic):
> + self.path = characteristic.path + '/desc' + str(index)
> + self.bus = bus
> + self.uuid = uuid
> + self.chrc = characteristic
> + dbus.service.Object.__init__(self, bus, self.path)
> +
> + def get_properties(self):
> + return {
> + GATT_DESC_IFACE: {
> + 'Characteristic': self.chrc.get_path(),
> + 'UUID': self.uuid,
> + }
> + }
> +
> + def get_path(self):
> + return dbus.ObjectPath(self.path)
> +
> + @dbus.service.method(DBUS_PROP_IFACE,
> + in_signature='s',
> + out_signature='a{sv}')
> + def GetAll(self, interface):
> + if interface != GATT_DESC_IFACE:
> + raise InvalidArgsException()
> +
> + return self.get_properties[GATT_CHRC_IFACE]
> +
> + @dbus.service.method(GATT_DESC_IFACE, out_signature='ay')
> + def ReadValue(self):
> + print 'Default ReadValue called, returning error'
> + raise NotSupportedException()
> +
> + @dbus.service.method(GATT_DESC_IFACE, in_signature='ay')
> + def WriteValue(self, value):
> + print 'Default WriteValue called, returning error'
> + raise NotSupportedException()
> +
> +
> +class HeartRateService(Service):
> + """
> + Fake Heart Rate Service that simulates a fake heart beat and control point
> + behavior.
> +
> + """
> + HR_UUID = '0000180d-0000-1000-8000-00805f9b34fb'
> +
> + def __init__(self, bus, index):
> + Service.__init__(self, bus, index, self.HR_UUID, True)
> + self.add_characteristic(HeartRateMeasurementChrc(bus, 0, self))
> + self.add_characteristic(BodySensorLocationChrc(bus, 1, self))
> + self.add_characteristic(HeartRateControlPointChrc(bus, 2, self))
> + self.energy_expended = 0
> +
> +
> +class HeartRateMeasurementChrc(Characteristic):
> + HR_MSRMT_UUID = '00002a37-0000-1000-8000-00805f9b34fb'
> +
> + def __init__(self, bus, index, service):
> + Characteristic.__init__(
> + self, bus, index,
> + self.HR_MSRMT_UUID,
> + ['notify'],
> + service)
> + self.notifying = False
> + self.hr_ee_count = 0
> +
> + def hr_msrmt_cb(self):
> + value = []
> + value.append(dbus.Byte(0x06))
> +
> + value.append(dbus.Byte(randint(90, 130)))
> +
> + if self.hr_ee_count % 10 == 0:
> + value[0] = dbus.Byte(value[0] | 0x08)
> + value.append(dbus.Byte(self.service.energy_expended & 0xff))
> + value.append(dbus.Byte((self.service.energy_expended >> 8) & 0xff))
> +
> + self.service.energy_expended = \
> + min(0xffff, self.service.energy_expended + 1)
> + self.hr_ee_count += 1
> +
> + print 'Updating value: ' + repr(value)
> +
> + self.PropertiesChanged(GATT_CHRC_IFACE, { 'Value': value }, [])
> +
> + return self.notifying
> +
> + def _update_hr_msrmt_simulation(self):
> + print 'Update HR Measurement Simulation'
> +
> + if not self.notifying:
> + return
> +
> + gobject.timeout_add(1000, self.hr_msrmt_cb)
> +
> + def StartNotify(self):
> + if self.notifying:
> + print 'Already notifying, nothing to do'
> + return
> +
> + self.notifying = True
> + self._update_hr_msrmt_simulation()
> +
> + def StopNotify(self):
> + if not self.notifying:
> + print 'Not notifying, nothing to do'
> + return
> +
> + self.notifying = False
> + self._update_hr_msrmt_simulation()
> +
> +
> +class BodySensorLocationChrc(Characteristic):
> + BODY_SNSR_LOC_UUID = '00002a38-0000-1000-8000-00805f9b34fb'
> +
> + def __init__(self, bus, index, service):
> + Characteristic.__init__(
> + self, bus, index,
> + self.BODY_SNSR_LOC_UUID,
> + ['read'],
> + service)
> +
> + def ReadValue(self):
> + # Return 'Chest' as the sensor location.
> + return [ 0x01 ]
> +
> +class HeartRateControlPointChrc(Characteristic):
> + HR_CTRL_PT_UUID = '00002a39-0000-1000-8000-00805f9b34fb'
> +
> + def __init__(self, bus, index, service):
> + Characteristic.__init__(
> + self, bus, index,
> + self.HR_CTRL_PT_UUID,
> + ['write'],
> + service)
> +
> + def WriteValue(self, value):
> + print 'Heart Rate Control Point WriteValue called'
> +
> + if len(value) != 1:
> + raise InvalidValueLengthException()
> +
> + byte = value[0]
> + print 'Control Point value: ' + repr(byte)
> +
> + if byte != 1:
> + raise FailedException("0x80")
> +
> + print 'Energy Expended field reset!'
> + self.service.energy_expended = 0
> +
> +
> +class BatteryService(Service):
> + """
> + Fake Battery service that emulates a draining battery.
> +
> + """
> + BATTERY_UUID = '180f'
> +
> + def __init__(self, bus, index):
> + Service.__init__(self, bus, index, self.BATTERY_UUID, True)
> + self.add_characteristic(BatteryLevelCharacteristic(bus, 0, self))
> +
> +
> +class BatteryLevelCharacteristic(Characteristic):
> + """
> + Fake Battery Level characteristic. The battery level is drained by 2 points
> + every 5 seconds.
> +
> + """
> + BATTERY_LVL_UUID = '2a19'
> +
> + def __init__(self, bus, index, service):
> + Characteristic.__init__(
> + self, bus, index,
> + self.BATTERY_LVL_UUID,
> + ['read', 'notify'],
> + service)
> + self.notifying = False
> + self.battery_lvl = 100
> + gobject.timeout_add(5000, self.drain_battery)
> +
> + def notify_battery_level(self):
> + if not self.notifying:
> + return
> + self.PropertiesChanged(
> + GATT_CHRC_IFACE,
> + { 'Value': [dbus.Byte(self.battery_lvl)] }, [])
> +
> + def drain_battery(self):
> + if self.battery_lvl > 0:
> + self.battery_lvl -= 2
> + if self.battery_lvl < 0:
> + self.battery_lvl = 0
> + print 'Battery Level drained: ' + repr(self.battery_lvl)
> + self.notify_battery_level()
> + return True
> +
> + def ReadValue(self):
> + print 'Battery Level read: ' + repr(self.battery_lvl)
> + return [dbus.Byte(self.battery_lvl)]
> +
> + def StartNotify(self):
> + if self.notifying:
> + print 'Already notifying, nothing to do'
> + return
> +
> + self.notifying = True
> + self.notify_battery_level()
> +
> + def StopNotify(self):
> + if not self.notifying:
> + print 'Not notifying, nothing to do'
> + return
> +
> + self.notifying = False
> +
> +
> +class TestService(Service):
> + """
> + Dummy test service that provides characteristics and descriptors that
> + exercise various API functionality.
> +
> + """
> + TEST_SVC_UUID = '12345678-1234-5678-1234-56789abcdef0'
> +
> + def __init__(self, bus, index):
> + Service.__init__(self, bus, index, self.TEST_SVC_UUID, False)
> + self.add_characteristic(TestCharacteristic(bus, 0, self))
> +
> +
> +class TestCharacteristic(Characteristic):
> + """
> + Dummy test characteristic. Allows writing arbitrary bytes to its value, and
> + contains "extended properties", as well as a test descriptor.
> +
> + """
> + TEST_CHRC_UUID = '12345678-1234-5678-1234-56789abcdef1'
> +
> + def __init__(self, bus, index, service):
> + Characteristic.__init__(
> + self, bus, index,
> + self.TEST_CHRC_UUID,
> + ['read', 'write', 'writable-auxiliaries'],
> + service)
> + self.value = []
> + self.add_descriptor(TestDescriptor(bus, 0, self))
> + self.add_descriptor(
> + CharacteristicUserDescriptionDescriptor(bus, 1, self))
> +
> + def ReadValue(self):
> + print 'TestCharacteristic Read: ' + repr(self.value)
> + return self.value
> +
> + def WriteValue(self, value):
> + print 'TestCharacteristic Write: ' + repr(value)
> + self.value = value
> +
> +
> +class TestDescriptor(Descriptor):
> + """
> + Dummy test descriptor. Returns a static value.
> +
> + """
> + TEST_DESC_UUID = '12345678-1234-5678-1234-56789abcdef2'
> +
> + def __init__(self, bus, index, characteristic):
> + Descriptor.__init__(
> + self, bus, index,
> + self.TEST_DESC_UUID,
> + characteristic)
> +
> + def ReadValue(self):
> + return [
> + dbus.Byte('T'), dbus.Byte('e'), dbus.Byte('s'), dbus.Byte('t')
> + ]
> +
> +
> +class CharacteristicUserDescriptionDescriptor(Descriptor):
> + """
> + Writable CUD descriptor.
> +
> + """
> + CUD_UUID = '2901'
> +
> + def __init__(self, bus, index, characteristic):
> + self.writable = 'writable-auxiliaries' in characteristic.flags
> + self.value = array.array('B', 'This is a characteristic for testing')
> + self.value = self.value.tolist()
> + Descriptor.__init__(
> + self, bus, index,
> + self.CUD_UUID,
> + characteristic)
> +
> + def ReadValue(self):
> + return self.value
> +
> + def WriteValue(self, value):
> + if not self.writable:
> + raise NotPermittedException()
> + self.value = value
> +
> +
> +def register_service_cb():
> + print 'GATT service registered'
> +
> +
> +def register_service_error_cb(error):
> + print 'Failed to register service: ' + str(error)
> + mainloop.quit()
> +
> +
> +def find_adapter(bus):
> + remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'),
> + DBUS_OM_IFACE)
> + objects = remote_om.GetManagedObjects()
> +
> + for o, props in objects.iteritems():
> + if props.has_key(GATT_MANAGER_IFACE):
> + return o
> +
> + return None
> +
> +def main():
> + global mainloop
> +
> + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
> +
> + bus = dbus.SystemBus()
> +
> + adapter = find_adapter(bus)
> + if not adapter:
> + print 'GattManager1 interface not found'
> + return
> +
> + service_manager = dbus.Interface(
> + bus.get_object(BLUEZ_SERVICE_NAME, adapter),
> + GATT_MANAGER_IFACE)
> +
> + hr_service = HeartRateService(bus, 0)
> + bat_service = BatteryService(bus, 1)
> + test_service = TestService(bus, 2)
> +
> + mainloop = gobject.MainLoop()
> +
> + service_manager.RegisterService(hr_service.get_path(), {},
> + reply_handler=register_service_cb,
> + error_handler=register_service_error_cb)
> + service_manager.RegisterService(bat_service.get_path(), {},
> + reply_handler=register_service_cb,
> + error_handler=register_service_error_cb)
> + service_manager.RegisterService(test_service.get_path(), {},
> + reply_handler=register_service_cb,
> + error_handler=register_service_error_cb)
> +
> + mainloop.run()
> +
> +if __name__ == '__main__':
> + main()
> diff --git a/tools/gatt-example b/tools/gatt-example
> deleted file mode 100644
> index a6f5cbe..0000000
> --- a/tools/gatt-example
> +++ /dev/null
> @@ -1,533 +0,0 @@
> -#!/usr/bin/python
> -
> -import dbus
> -import dbus.exceptions
> -import dbus.mainloop.glib
> -import dbus.service
> -
> -import array
> -import gobject
> -
> -from random import randint
> -
> -mainloop = None
> -
> -BLUEZ_SERVICE_NAME = 'org.bluez'
> -GATT_MANAGER_IFACE = 'org.bluez.GattManager1'
> -DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
> -DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
> -
> -GATT_SERVICE_IFACE = 'org.bluez.GattService1'
> -GATT_CHRC_IFACE = 'org.bluez.GattCharacteristic1'
> -GATT_DESC_IFACE = 'org.bluez.GattDescriptor1'
> -
> -class InvalidArgsException(dbus.exceptions.DBusException):
> - _dbus_error_name = 'org.freedesktop.DBus.Error.InvalidArgs'
> -
> -class NotSupportedException(dbus.exceptions.DBusException):
> - _dbus_error_name = 'org.bluez.Error.NotSupported'
> -
> -class NotPermittedException(dbus.exceptions.DBusException):
> - _dbus_error_name = 'org.bluez.Error.NotPermitted'
> -
> -class InvalidValueLengthException(dbus.exceptions.DBusException):
> - _dbus_error_name = 'org.bluez.Error.InvalidValueLength'
> -
> -class FailedException(dbus.exceptions.DBusException):
> - _dbus_error_name = 'org.bluez.Error.Failed'
> -
> -
> -class Service(dbus.service.Object):
> - PATH_BASE = '/org/bluez/example/service'
> -
> - def __init__(self, bus, index, uuid, primary):
> - self.path = self.PATH_BASE + str(index)
> - self.bus = bus
> - self.uuid = uuid
> - self.primary = primary
> - self.characteristics = []
> - dbus.service.Object.__init__(self, bus, self.path)
> -
> - def get_properties(self):
> - return {
> - GATT_SERVICE_IFACE: {
> - 'UUID': self.uuid,
> - 'Primary': self.primary,
> - 'Characteristics': dbus.Array(
> - self.get_characteristic_paths(),
> - signature='o')
> - }
> - }
> -
> - def get_path(self):
> - return dbus.ObjectPath(self.path)
> -
> - def add_characteristic(self, characteristic):
> - self.characteristics.append(characteristic)
> -
> - def get_characteristic_paths(self):
> - result = []
> - for chrc in self.characteristics:
> - result.append(chrc.get_path())
> - return result
> -
> - def get_characteristics(self):
> - return self.characteristics
> -
> - @dbus.service.method(DBUS_PROP_IFACE,
> - in_signature='s',
> - out_signature='a{sv}')
> - def GetAll(self, interface):
> - if interface != GATT_SERVICE_IFACE:
> - raise InvalidArgsException()
> -
> - return self.get_properties[GATT_SERVICE_IFACE]
> -
> - @dbus.service.method(DBUS_OM_IFACE, out_signature='a{oa{sa{sv}}}')
> - def GetManagedObjects(self):
> - response = {}
> - print 'GetManagedObjects'
> -
> - response[self.get_path()] = self.get_properties()
> - chrcs = self.get_characteristics()
> - for chrc in chrcs:
> - response[chrc.get_path()] = chrc.get_properties()
> - descs = chrc.get_descriptors()
> - for desc in descs:
> - response[desc.get_path()] = desc.get_properties()
> -
> - return response
> -
> -
> -class Characteristic(dbus.service.Object):
> - def __init__(self, bus, index, uuid, flags, service):
> - self.path = service.path + '/char' + str(index)
> - self.bus = bus
> - self.uuid = uuid
> - self.service = service
> - self.flags = flags
> - self.descriptors = []
> - dbus.service.Object.__init__(self, bus, self.path)
> -
> - def get_properties(self):
> - return {
> - GATT_CHRC_IFACE: {
> - 'Service': self.service.get_path(),
> - 'UUID': self.uuid,
> - 'Flags': self.flags,
> - 'Descriptors': dbus.Array(
> - self.get_descriptor_paths(),
> - signature='o')
> - }
> - }
> -
> - def get_path(self):
> - return dbus.ObjectPath(self.path)
> -
> - def add_descriptor(self, descriptor):
> - self.descriptors.append(descriptor)
> -
> - def get_descriptor_paths(self):
> - result = []
> - for desc in self.descriptors:
> - result.append(desc.get_path())
> - return result
> -
> - def get_descriptors(self):
> - return self.descriptors
> -
> - @dbus.service.method(DBUS_PROP_IFACE,
> - in_signature='s',
> - out_signature='a{sv}')
> - def GetAll(self, interface):
> - if interface != GATT_CHRC_IFACE:
> - raise InvalidArgsException()
> -
> - return self.get_properties[GATT_CHRC_IFACE]
> -
> - @dbus.service.method(GATT_CHRC_IFACE, out_signature='ay')
> - def ReadValue(self):
> - print 'Default ReadValue called, returning error'
> - raise NotSupportedException()
> -
> - @dbus.service.method(GATT_CHRC_IFACE, in_signature='ay')
> - def WriteValue(self, value):
> - print 'Default WriteValue called, returning error'
> - raise NotSupportedException()
> -
> - @dbus.service.method(GATT_CHRC_IFACE)
> - def StartNotify(self):
> - print 'Default StartNotify called, returning error'
> - raise NotSupportedException()
> -
> - @dbus.service.method(GATT_CHRC_IFACE)
> - def StopNotify(self):
> - print 'Default StopNotify called, returning error'
> - raise NotSupportedException()
> -
> - @dbus.service.signal(DBUS_PROP_IFACE,
> - signature='sa{sv}as')
> - def PropertiesChanged(self, interface, changed, invalidated):
> - pass
> -
> -
> -class Descriptor(dbus.service.Object):
> - def __init__(self, bus, index, uuid, characteristic):
> - self.path = characteristic.path + '/desc' + str(index)
> - self.bus = bus
> - self.uuid = uuid
> - self.chrc = characteristic
> - dbus.service.Object.__init__(self, bus, self.path)
> -
> - def get_properties(self):
> - return {
> - GATT_DESC_IFACE: {
> - 'Characteristic': self.chrc.get_path(),
> - 'UUID': self.uuid,
> - }
> - }
> -
> - def get_path(self):
> - return dbus.ObjectPath(self.path)
> -
> - @dbus.service.method(DBUS_PROP_IFACE,
> - in_signature='s',
> - out_signature='a{sv}')
> - def GetAll(self, interface):
> - if interface != GATT_DESC_IFACE:
> - raise InvalidArgsException()
> -
> - return self.get_properties[GATT_CHRC_IFACE]
> -
> - @dbus.service.method(GATT_DESC_IFACE, out_signature='ay')
> - def ReadValue(self):
> - print 'Default ReadValue called, returning error'
> - raise NotSupportedException()
> -
> - @dbus.service.method(GATT_DESC_IFACE, in_signature='ay')
> - def WriteValue(self, value):
> - print 'Default WriteValue called, returning error'
> - raise NotSupportedException()
> -
> -
> -class HeartRateService(Service):
> - """
> - Fake Heart Rate Service that simulates a fake heart beat and control point
> - behavior.
> -
> - """
> - HR_UUID = '0000180d-0000-1000-8000-00805f9b34fb'
> -
> - def __init__(self, bus, index):
> - Service.__init__(self, bus, index, self.HR_UUID, True)
> - self.add_characteristic(HeartRateMeasurementChrc(bus, 0, self))
> - self.add_characteristic(BodySensorLocationChrc(bus, 1, self))
> - self.add_characteristic(HeartRateControlPointChrc(bus, 2, self))
> - self.energy_expended = 0
> -
> -
> -class HeartRateMeasurementChrc(Characteristic):
> - HR_MSRMT_UUID = '00002a37-0000-1000-8000-00805f9b34fb'
> -
> - def __init__(self, bus, index, service):
> - Characteristic.__init__(
> - self, bus, index,
> - self.HR_MSRMT_UUID,
> - ['notify'],
> - service)
> - self.notifying = False
> - self.hr_ee_count = 0
> -
> - def hr_msrmt_cb(self):
> - value = []
> - value.append(dbus.Byte(0x06))
> -
> - value.append(dbus.Byte(randint(90, 130)))
> -
> - if self.hr_ee_count % 10 == 0:
> - value[0] = dbus.Byte(value[0] | 0x08)
> - value.append(dbus.Byte(self.service.energy_expended & 0xff))
> - value.append(dbus.Byte((self.service.energy_expended >> 8) & 0xff))
> -
> - self.service.energy_expended = \
> - min(0xffff, self.service.energy_expended + 1)
> - self.hr_ee_count += 1
> -
> - print 'Updating value: ' + repr(value)
> -
> - self.PropertiesChanged(GATT_CHRC_IFACE, { 'Value': value }, [])
> -
> - return self.notifying
> -
> - def _update_hr_msrmt_simulation(self):
> - print 'Update HR Measurement Simulation'
> -
> - if not self.notifying:
> - return
> -
> - gobject.timeout_add(1000, self.hr_msrmt_cb)
> -
> - def StartNotify(self):
> - if self.notifying:
> - print 'Already notifying, nothing to do'
> - return
> -
> - self.notifying = True
> - self._update_hr_msrmt_simulation()
> -
> - def StopNotify(self):
> - if not self.notifying:
> - print 'Not notifying, nothing to do'
> - return
> -
> - self.notifying = False
> - self._update_hr_msrmt_simulation()
> -
> -
> -class BodySensorLocationChrc(Characteristic):
> - BODY_SNSR_LOC_UUID = '00002a38-0000-1000-8000-00805f9b34fb'
> -
> - def __init__(self, bus, index, service):
> - Characteristic.__init__(
> - self, bus, index,
> - self.BODY_SNSR_LOC_UUID,
> - ['read'],
> - service)
> -
> - def ReadValue(self):
> - # Return 'Chest' as the sensor location.
> - return [ 0x01 ]
> -
> -class HeartRateControlPointChrc(Characteristic):
> - HR_CTRL_PT_UUID = '00002a39-0000-1000-8000-00805f9b34fb'
> -
> - def __init__(self, bus, index, service):
> - Characteristic.__init__(
> - self, bus, index,
> - self.HR_CTRL_PT_UUID,
> - ['write'],
> - service)
> -
> - def WriteValue(self, value):
> - print 'Heart Rate Control Point WriteValue called'
> -
> - if len(value) != 1:
> - raise InvalidValueLengthException()
> -
> - byte = value[0]
> - print 'Control Point value: ' + repr(byte)
> -
> - if byte != 1:
> - raise FailedException("0x80")
> -
> - print 'Energy Expended field reset!'
> - self.service.energy_expended = 0
> -
> -
> -class BatteryService(Service):
> - """
> - Fake Battery service that emulates a draining battery.
> -
> - """
> - BATTERY_UUID = '180f'
> -
> - def __init__(self, bus, index):
> - Service.__init__(self, bus, index, self.BATTERY_UUID, True)
> - self.add_characteristic(BatteryLevelCharacteristic(bus, 0, self))
> -
> -
> -class BatteryLevelCharacteristic(Characteristic):
> - """
> - Fake Battery Level characteristic. The battery level is drained by 2 points
> - every 5 seconds.
> -
> - """
> - BATTERY_LVL_UUID = '2a19'
> -
> - def __init__(self, bus, index, service):
> - Characteristic.__init__(
> - self, bus, index,
> - self.BATTERY_LVL_UUID,
> - ['read', 'notify'],
> - service)
> - self.notifying = False
> - self.battery_lvl = 100
> - gobject.timeout_add(5000, self.drain_battery)
> -
> - def notify_battery_level(self):
> - if not self.notifying:
> - return
> - self.PropertiesChanged(
> - GATT_CHRC_IFACE,
> - { 'Value': [dbus.Byte(self.battery_lvl)] }, [])
> -
> - def drain_battery(self):
> - if self.battery_lvl > 0:
> - self.battery_lvl -= 2
> - if self.battery_lvl < 0:
> - self.battery_lvl = 0
> - print 'Battery Level drained: ' + repr(self.battery_lvl)
> - self.notify_battery_level()
> - return True
> -
> - def ReadValue(self):
> - print 'Battery Level read: ' + repr(self.battery_lvl)
> - return [dbus.Byte(self.battery_lvl)]
> -
> - def StartNotify(self):
> - if self.notifying:
> - print 'Already notifying, nothing to do'
> - return
> -
> - self.notifying = True
> - self.notify_battery_level()
> -
> - def StopNotify(self):
> - if not self.notifying:
> - print 'Not notifying, nothing to do'
> - return
> -
> - self.notifying = False
> -
> -
> -class TestService(Service):
> - """
> - Dummy test service that provides characteristics and descriptors that
> - exercise various API functionality.
> -
> - """
> - TEST_SVC_UUID = '12345678-1234-5678-1234-56789abcdef0'
> -
> - def __init__(self, bus, index):
> - Service.__init__(self, bus, index, self.TEST_SVC_UUID, False)
> - self.add_characteristic(TestCharacteristic(bus, 0, self))
> -
> -
> -class TestCharacteristic(Characteristic):
> - """
> - Dummy test characteristic. Allows writing arbitrary bytes to its value, and
> - contains "extended properties", as well as a test descriptor.
> -
> - """
> - TEST_CHRC_UUID = '12345678-1234-5678-1234-56789abcdef1'
> -
> - def __init__(self, bus, index, service):
> - Characteristic.__init__(
> - self, bus, index,
> - self.TEST_CHRC_UUID,
> - ['read', 'write', 'writable-auxiliaries'],
> - service)
> - self.value = []
> - self.add_descriptor(TestDescriptor(bus, 0, self))
> - self.add_descriptor(
> - CharacteristicUserDescriptionDescriptor(bus, 1, self))
> -
> - def ReadValue(self):
> - print 'TestCharacteristic Read: ' + repr(self.value)
> - return self.value
> -
> - def WriteValue(self, value):
> - print 'TestCharacteristic Write: ' + repr(value)
> - self.value = value
> -
> -
> -class TestDescriptor(Descriptor):
> - """
> - Dummy test descriptor. Returns a static value.
> -
> - """
> - TEST_DESC_UUID = '12345678-1234-5678-1234-56789abcdef2'
> -
> - def __init__(self, bus, index, characteristic):
> - Descriptor.__init__(
> - self, bus, index,
> - self.TEST_DESC_UUID,
> - characteristic)
> -
> - def ReadValue(self):
> - return [
> - dbus.Byte('T'), dbus.Byte('e'), dbus.Byte('s'), dbus.Byte('t')
> - ]
> -
> -
> -class CharacteristicUserDescriptionDescriptor(Descriptor):
> - """
> - Writable CUD descriptor.
> -
> - """
> - CUD_UUID = '2901'
> -
> - def __init__(self, bus, index, characteristic):
> - self.writable = 'writable-auxiliaries' in characteristic.flags
> - self.value = array.array('B', 'This is a characteristic for testing')
> - self.value = self.value.tolist()
> - Descriptor.__init__(
> - self, bus, index,
> - self.CUD_UUID,
> - characteristic)
> -
> - def ReadValue(self):
> - return self.value
> -
> - def WriteValue(self, value):
> - if not self.writable:
> - raise NotPermittedException()
> - self.value = value
> -
> -
> -def register_service_cb():
> - print 'GATT service registered'
> -
> -
> -def register_service_error_cb(error):
> - print 'Failed to register service: ' + str(error)
> - mainloop.quit()
> -
> -
> -def find_adapter(bus):
> - remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'),
> - DBUS_OM_IFACE)
> - objects = remote_om.GetManagedObjects()
> -
> - for o, props in objects.iteritems():
> - if props.has_key(GATT_MANAGER_IFACE):
> - return o
> -
> - return None
> -
> -def main():
> - global mainloop
> -
> - dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
> -
> - bus = dbus.SystemBus()
> -
> - adapter = find_adapter(bus)
> - if not adapter:
> - print 'GattManager1 interface not found'
> - return
> -
> - service_manager = dbus.Interface(
> - bus.get_object(BLUEZ_SERVICE_NAME, adapter),
> - GATT_MANAGER_IFACE)
> -
> - hr_service = HeartRateService(bus, 0)
> - bat_service = BatteryService(bus, 1)
> - test_service = TestService(bus, 2)
> -
> - mainloop = gobject.MainLoop()
> -
> - service_manager.RegisterService(hr_service.get_path(), {},
> - reply_handler=register_service_cb,
> - error_handler=register_service_error_cb)
> - service_manager.RegisterService(bat_service.get_path(), {},
> - reply_handler=register_service_cb,
> - error_handler=register_service_error_cb)
> - service_manager.RegisterService(test_service.get_path(), {},
> - reply_handler=register_service_cb,
> - error_handler=register_service_error_cb)
> -
> - mainloop.run()
> -
> -if __name__ == '__main__':
> - main()
> --
> 2.2.0.rc0.207.ga3a616c
>
Pushed.
Thanks,
Arman