path: root/test/example-adv-monitor

#!/usr/bin/python
# SPDX-License-Identifier: LGPL-2.1-or-later

import argparse
import dbus
import dbus.mainloop.glib
import dbus.service
import json
import time

from threading import Thread

try:
    from gi.repository import GObject  # python3
except ImportError:
    import gobject as GObject  # python2

DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'

BLUEZ_SERVICE_NAME = 'org.bluez'

ADV_MONITOR_MANAGER_IFACE = 'org.bluez.AdvertisementMonitorManager1'
ADV_MONITOR_IFACE = 'org.bluez.AdvertisementMonitor1'
ADV_MONITOR_APP_BASE_PATH = '/org/bluez/example/adv_monitor_app'


class AdvMonitor(dbus.service.Object):

    # Indexes of the Monitor object parameters in a monitor data list.
    MONITOR_TYPE = 0
    RSSI_FILTER = 1
    PATTERNS = 2

    # Indexes of the RSSI filter parameters in a monitor data list.
    RSSI_H_THRESH = 0
    RSSI_H_TIMEOUT = 1
    RSSI_L_THRESH = 2
    RSSI_L_TIMEOUT = 3

    # Indexes of the Patterns filter parameters in a monitor data list.
    PATTERN_START_POS = 0
    PATTERN_AD_TYPE = 1
    PATTERN_DATA = 2

    def __init__(self, bus, app_path, monitor_id, monitor_data):
        self.path = app_path + '/monitor' + str(monitor_id)
        self.bus = bus

        self._set_type(monitor_data[self.MONITOR_TYPE])
        self._set_rssi(monitor_data[self.RSSI_FILTER])
        self._set_patterns(monitor_data[self.PATTERNS])

        super(AdvMonitor, self).__init__(self.bus, self.path)


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


    def get_properties(self):
        properties = dict()
        properties['Type'] = dbus.String(self.monitor_type)
        properties['RSSIHighThreshold'] = dbus.Int16(self.rssi_h_thresh)
        properties['RSSIHighTimeout'] = dbus.UInt16(self.rssi_h_timeout)
        properties['RSSILowThreshold'] = dbus.Int16(self.rssi_l_thresh)
        properties['RSSILowTimeout'] = dbus.UInt16(self.rssi_l_timeout)
        properties['Patterns'] = dbus.Array(self.patterns, signature='(yyay)')
        return {ADV_MONITOR_IFACE: properties}


    def _set_type(self, monitor_type):
        self.monitor_type = monitor_type


    def _set_rssi(self, rssi):
        self.rssi_h_thresh = rssi[self.RSSI_H_THRESH]
        self.rssi_h_timeout = rssi[self.RSSI_H_TIMEOUT]
        self.rssi_l_thresh = rssi[self.RSSI_L_THRESH]
        self.rssi_l_timeout = rssi[self.RSSI_L_TIMEOUT]


    def _set_patterns(self, patterns):
        self.patterns = []
        for pattern in patterns:
            start_pos = dbus.Byte(pattern[self.PATTERN_START_POS])
            ad_type = dbus.Byte(pattern[self.PATTERN_AD_TYPE])
            ad_data = []
            for byte in pattern[self.PATTERN_DATA]:
                ad_data.append(dbus.Byte(byte))
            adv_pattern = dbus.Struct((start_pos, ad_type, ad_data),
                                      signature='yyay')
            self.patterns.append(adv_pattern)


    def remove_monitor(self):
        self.remove_from_connection()


    @dbus.service.method(DBUS_PROP_IFACE,
                         in_signature='s',
                         out_signature='a{sv}')
    def GetAll(self, interface):
        print('{}: {} GetAll'.format(self.path, interface))
        if interface != ADV_MONITOR_IFACE:
            print('{}: GetAll: Invalid arg {}'.format(self.path, interface))
            return {}

        return self.get_properties()[ADV_MONITOR_IFACE]


    @dbus.service.method(ADV_MONITOR_IFACE,
                         in_signature='',
                         out_signature='')
    def Activate(self):
        print('{}: Monitor Activated'.format(self.path))


    @dbus.service.method(ADV_MONITOR_IFACE,
                         in_signature='',
                         out_signature='')
    def Release(self):
        print('{}: Monitor Released'.format(self.path))


    @dbus.service.method(ADV_MONITOR_IFACE,
                         in_signature='o',
                         out_signature='')
    def DeviceFound(self, device):
        print('{}: {} Device Found'.format(self.path, device))


    @dbus.service.method(ADV_MONITOR_IFACE,
                         in_signature='o',
                         out_signature='')
    def DeviceLost(self, device):
        print('{}: {} Device Lost'.format(self.path, device))


class AdvMonitorApp(dbus.service.Object):

    def __init__(self, bus, advmon_manager, app_id):
        self.bus = bus
        self.advmon_mgr = advmon_manager
        self.app_path = ADV_MONITOR_APP_BASE_PATH + str(app_id)

        self.monitors = dict()

        super(AdvMonitorApp, self).__init__(self.bus, self.app_path)


    def get_app_path(self):
        return dbus.ObjectPath(self.app_path)


    def add_monitor(self, monitor_data):
        monitor_id = 0
        while monitor_id in self.monitors:
            monitor_id += 1

        monitor = AdvMonitor(self.bus, self.app_path, monitor_id, monitor_data)

        # Emit the InterfacesAdded signal once the Monitor object is created.
        self.InterfacesAdded(monitor.get_path(), monitor.get_properties())

        self.monitors[monitor_id] = monitor

        return monitor_id


    def remove_monitor(self, monitor_id):
        monitor = self.monitors.pop(monitor_id, None)
        if not monitor:
            return False

        # Emit the InterfacesRemoved signal before removing the Monitor object.
        self.InterfacesRemoved(monitor.get_path(),
                               monitor.get_properties().keys())

        monitor.remove_monitor()

        return True


    def register_app(self):
        self.register_successful = None

        def register_cb():
            print('{}: RegisterMonitor successful'.format(self.app_path))
            self.register_successful = True

        def register_error_cb(error):
            print('{}: RegisterMonitor failed: {}'.format(self.app_path,
                                                          str(error)))
            self.register_successful = False

        self.advmon_mgr.RegisterMonitor(self.get_app_path(),
                                        reply_handler=register_cb,
                                        error_handler=register_error_cb)

        # Wait for the reply.
        while self.register_successful is None:
            pass

        return self.register_successful


    def unregister_app(self):
        self.unregister_successful = None

        def unregister_cb():
            print('{}: UnregisterMonitor successful'.format(self.app_path))
            self.unregister_successful = True

        def unregister_error_cb(error):
            print('{}: UnregisterMonitor failed: {}'.format(self.app_path,
                                                            str(error)))
            self.unregister_successful = False

        self.advmon_mgr.UnregisterMonitor(self.get_app_path(),
                                          reply_handler=unregister_cb,
                                          error_handler=unregister_error_cb)

        # Wait for the reply.
        while self.unregister_successful is None:
            pass

        return self.unregister_successful


    @dbus.service.method(DBUS_OM_IFACE, out_signature='a{oa{sa{sv}}}')
    def GetManagedObjects(self):
        print('{}: GetManagedObjects'.format(self.app_path))
        objects = dict()
        for monitor_id in self.monitors:
            monitor = self.monitors[monitor_id]
            objects[monitor.get_path()] = monitor.get_properties()

        return objects


    @dbus.service.signal(DBUS_OM_IFACE, signature='oa{sa{sv}}')
    def InterfacesAdded(self, object_path, interfaces_and_properties):
        # Invoking this method emits the InterfacesAdded signal,
        # nothing needs to be done here.
        return


    @dbus.service.signal(DBUS_OM_IFACE, signature='oas')
    def InterfacesRemoved(self, object_path, interfaces):
        # Invoking this method emits the InterfacesRemoved signal,
        # nothing needs to be done here.
        return


def read_adapter_supported_monitor_types(adapter_props):
    types = json.dumps(adapter_props.Get(ADV_MONITOR_MANAGER_IFACE,
                       'SupportedMonitorTypes',
                       dbus_interface=DBUS_PROP_IFACE))
    return json.loads(types)


def read_adapter_supported_monitor_features(adapter_props):
    features = json.dumps(adapter_props.Get(ADV_MONITOR_MANAGER_IFACE,
                          'SupportedFeatures',
                          dbus_interface=DBUS_PROP_IFACE))
    return json.loads(features)


def print_supported_types_and_features(adapter_props):
    supported_types = read_adapter_supported_monitor_types(adapter_props)
    for supported_type in supported_types:
        print(supported_type)

    supported_features = read_adapter_supported_monitor_features(adapter_props)
    for supported_feature in supported_features:
        print(supported_feature)


def find_advmon_mgr(bus, adapter):
    return dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, adapter),
                          ADV_MONITOR_MANAGER_IFACE)


def find_adapter(bus):
    remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'),
                                DBUS_OM_IFACE)
    objects = remote_om.GetManagedObjects()

    adapter = None
    adapter_props = None

    for o, props in objects.items():
        if ADV_MONITOR_MANAGER_IFACE in props:
            adapter = o
            break

    if adapter:
        # Turn on the bluetooth adapter.
        adapter_props = dbus.Interface(
                                bus.get_object(BLUEZ_SERVICE_NAME, adapter),
                                DBUS_PROP_IFACE)
        adapter_props.Set('org.bluez.Adapter1', 'Powered', dbus.Boolean(1))

    return adapter, adapter_props


def test(bus, mainloop, advmon_mgr, app_id):
    # Create an App instance.
    app = AdvMonitorApp(bus, advmon_mgr, app_id)

    # Create two monitor objects before registering the app. No Activate() or
    # Release() should get called yet as the app is not registered.
    data0 = [
        'invalid_patterns',
        [-50, 1, -70, 1],
        [[0, 0x03, [0x12, 0x18]]] # Service Class UUID is 0x1812 (HOG)
    ]
    data1 = [
        'or_patterns',
        [127, 0, 127, 0],
        [[5, 0x09, [ord('_')]]] # 5th character of the Local Name is '_'
    ]
    monitor0 = app.add_monitor(data0)
    monitor1 = app.add_monitor(data1)

    # Register the app root path to expose advertisement monitors.
    # Release() should get called on monitor0 - incorrect monitor type.
    # Activate() should get called on monitor1.
    ret = app.register_app()
    if not ret:
        print('RegisterMonitor failed.')
        mainloop.quit()
        exit(-1)

    # Create two more monitor objects.
    # Release() should get called on monitor2 - incorrect RSSI Filter values.
    # Activate() should get called on monitor3.
    data2 = [
        'or_patterns',
        [-50, 1, -30, 1],
        [[0, 0x19, [0xC2, 0x03]]] # Appearance is 0xC203 (Mouse)
    ]
    data3 = [
        'or_patterns',
        [-50, 1, -70, 1],
        [[0, 0x03, [0x12, 0x18]], [0, 0x19, [0xC2, 0x03]]]
    ]
    monitor2 = app.add_monitor(data2)
    monitor3 = app.add_monitor(data3)

    # Run until user hits the 'Enter' key. If any peer device is advertising
    # during this time, DeviceFound() should get triggered for monitors
    # matching the advertisements.
    raw_input('Press "Enter" key to quit...\n')

    # Remove a monitor. DeviceFound() for this monitor should not get
    # triggered any more.
    app.remove_monitor(monitor1)

    # Unregister the app. Release() should get invoked on active monitors,
    # monitor3 in this case.
    app.unregister_app()

    mainloop.quit()


def main(app_id):
    # Initialize threads in gobject/dbus-glib before creating local threads.
    GObject.threads_init()
    dbus.mainloop.glib.threads_init()

    # Arrange for the GLib main loop to be the default.
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

    bus = dbus.SystemBus()
    mainloop = GObject.MainLoop()

    # Find bluetooth adapter and power it on.
    adapter, adapter_props = find_adapter(bus)
    if not adapter or not adapter_props:
        print('Bluetooth adapter not found.')
        exit(-1)

    # Read supported types and find AdvertisementMonitorManager1 interface.
    print_supported_types_and_features(adapter_props)
    advmon_mgr = find_advmon_mgr(bus, adapter)
    if not advmon_mgr :
        print('AdvertisementMonitorManager1 interface not found.')
        exit(-1)

    Thread(target=test, args=(bus, mainloop, advmon_mgr, app_id)).start()

    mainloop.run() # blocks until mainloop.quit() is called


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--app_id', default=0, type=int, help='use this App-ID '
                        'for creating dbus objects (default: 0)')
    args = parser.parse_args()

    main(args.app_id)