Blob: test-gatt-profile

Blob id: 64ff3e5f9502dda42d3a503942d75c6edb0116a7

Size: 3.4 KB

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
#!/usr/bin/env python3
# SPDX-License-Identifier: LGPL-2.1-or-later

from __future__ import absolute_import, print_function, unicode_literals

from optparse import OptionParser, make_option
import os
import sys
import uuid
import dbus
import dbus.service
import dbus.mainloop.glib
try:
  from gi.repository import GObject
except ImportError:
  import gobject as GObject
import bluezutils

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_PROFILE_IFACE = 'org.bluez.GattProfile1'


class InvalidArgsException(dbus.exceptions.DBusException):
    _dbus_error_name = 'org.freedesktop.DBus.Error.InvalidArgs'


class Application(dbus.service.Object):
    def __init__(self, bus):
        self.path = '/'
        self.profiles = []
        dbus.service.Object.__init__(self, bus, self.path)

    def get_path(self):
        return dbus.ObjectPath(self.path)

    def add_profile(self, profile):
        self.profiles.append(profile)

    @dbus.service.method(DBUS_OM_IFACE, out_signature='a{oa{sa{sv}}}')
    def GetManagedObjects(self):
        response = {}
        print('GetManagedObjects')

        for profile in self.profiles:
            response[profile.get_path()] = profile.get_properties()

        return response


class Profile(dbus.service.Object):
    PATH_BASE = '/org/bluez/example/profile'

    def __init__(self, bus, uuids):
        self.path = self.PATH_BASE
        self.bus = bus
        self.uuids = uuids
        dbus.service.Object.__init__(self, bus, self.path)

    def get_properties(self):
        return {
            GATT_PROFILE_IFACE: {
                'UUIDs': self.uuids,
            }
        }

    def get_path(self):
        return dbus.ObjectPath(self.path)

    @dbus.service.method(GATT_PROFILE_IFACE,
                        in_signature="",
                        out_signature="")
    def Release(self):
        print("Release")
        mainloop.quit()

    @dbus.service.method(DBUS_PROP_IFACE,
                         in_signature='s',
                         out_signature='a{sv}')
    def GetAll(self, interface):
        if interface != GATT_PROFILE_IFACE:
            raise InvalidArgsException()

        return self.get_properties[GATT_PROFILE_IFACE]


def register_app_cb():
    print('GATT application registered')


def register_app_error_cb(error):
    print('Failed to register application: ' + str(error))
    mainloop.quit()

if __name__ == '__main__':
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

    bus = dbus.SystemBus()

    path = bluezutils.find_adapter().object_path

    manager = dbus.Interface(bus.get_object("org.bluez", path),
                            GATT_MANAGER_IFACE)

    option_list = [make_option("-u", "--uuid", action="store",
                                type="string", dest="uuid",
                                default=None),
    ]

    opts = dbus.Dictionary({}, signature='sv')

    parser = OptionParser(option_list=option_list)

    (options, args) = parser.parse_args()

    mainloop = GObject.MainLoop()

    if not options.uuid:
        options.uuid = str(uuid.uuid4())

    app = Application(bus)
    profile = Profile(bus, [options.uuid])
    app.add_profile(profile)
    manager.RegisterApplication(app.get_path(), {},
                                reply_handler=register_app_cb,
                                error_handler=register_app_error_cb)

    mainloop.run()