bluez/test/example-battery-provider

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

import dbus
import dbus.exceptions
import dbus.mainloop.glib
import dbus.service

try:
  from gi.repository import GObject
except ImportError:
  import gobject as GObject
import sys

mainloop = None
app = None
bus = None

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

BATTERY_PROVIDER_MANAGER_IFACE = 'org.bluez.BatteryProviderManager1'
BATTERY_PROVIDER_IFACE = 'org.bluez.BatteryProvider1'
BATTERY_PROVIDER_PATH = '/path/to/provider'

BATTERY_PATH1 = '11_11_11_11_11_11'
BATTERY_PATH2 = '22_22_22_22_22_22'
BATTERY_PATH3 = '33_33_33_33_33_33'

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


class Application(dbus.service.Object):
    def __init__(self, bus):
        self.path = BATTERY_PROVIDER_PATH
        self.services = []
        self.batteries = []
        dbus.service.Object.__init__(self, bus, self.path)

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

    def add_battery(self, battery):
        self.batteries.append(battery)
        self.InterfacesAdded(battery.get_path(), battery.get_properties())
        GObject.timeout_add(1000, drain_battery, battery)

    def remove_battery(self, battery):
        self.batteries.remove(battery)
        self.InterfacesRemoved(battery.get_path(), [BATTERY_PROVIDER_IFACE])

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

        for battery in self.batteries:
            response[battery.get_path()] = battery.get_properties()

        return response

    @dbus.service.signal(DBUS_OM_IFACE, signature='oa{sa{sv}}')
    def InterfacesAdded(self, object_path, interfaces_and_properties):
        return

    @dbus.service.signal(DBUS_OM_IFACE, signature='oas')
    def InterfacesRemoved(self, object_path, interfaces):
        return


class Battery(dbus.service.Object):
    """
    org.bluez.BatteryProvider1 interface implementation
    """
    def __init__(self, bus, dev, percentage, source = None):
        self.path = BATTERY_PROVIDER_PATH + '/dev_' + dev
        self.dev_path = '/org/bluez/hci0/dev_' + dev
        self.bus = bus
        self.percentage = percentage
        self.source = source
        dbus.service.Object.__init__(self, bus, self.path)

    def get_battery_properties(self):
        properties = {}
        if self.percentage != None:
            properties['Percentage'] = dbus.Byte(self.percentage)
        if self.source != None:
            properties['Source'] = self.source
        properties['Device'] = dbus.ObjectPath(self.dev_path)
        return properties

    def get_properties(self):
        return { BATTERY_PROVIDER_IFACE: self.get_battery_properties() }

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

    def set_percentage(self, percentage):
        if percentage < 0 or percentage > 100:
            print('percentage not valid')
            return

        self.percentage = percentage
        print('battery %s percentage %d' % (self.path, self.percentage))
        self.PropertiesChanged(
                BATTERY_PROVIDER_IFACE, self.get_battery_properties())

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

        return self.get_properties()[BATTERY_PROVIDER_IFACE]

    @dbus.service.signal(DBUS_PROP_IFACE, signature='sa{sv}')
    def PropertiesChanged(self, interface, properties):
        return


def add_late_battery():
    app.add_battery(Battery(bus, BATTERY_PATH3, 70, 'Protocol 2'))


def drain_battery(battery):
    new_percentage = 100
    if battery.percentage != None:
        new_percentage = battery.percentage - 5
        if new_percentage < 0:
            new_percentage = 0

    battery.set_percentage(new_percentage)

    if new_percentage <= 0:
        return False

    return True

def register_provider_cb():
    print('Battery Provider registered')

    # Battery added early right after RegisterBatteryProvider succeeds
    app.add_battery(Battery(bus, BATTERY_PATH2, None))
    # Battery added later
    GObject.timeout_add(5000, add_late_battery)


def register_provider_error_cb(error):
    print('Failed to register Battery Provider: ' + str(error))
    mainloop.quit()


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

    for o, props in objects.items():
        if BATTERY_PROVIDER_MANAGER_IFACE in props.keys():
            return o

    return None


def unregister_provider_cb():
    print('Battery Provider unregistered')


def unregister_provider_error_cb(error):
    print('Failed to unregister Battery Provider: ' + str(error))


def unregister_battery_provider(battery_provider_manager):
    battery_provider_manager.UnregisterBatteryProvider(BATTERY_PROVIDER_PATH,
                                    reply_handler=unregister_provider_cb,
                                    error_handler=unregister_provider_error_cb)


def remove_battery(app, battery):
    app.remove_battery(battery)


"""
Simulates an application registering to BlueZ as a Battery Provider providing
fake batteries drained periodically.
"""
def main():
    global mainloop, bus, app

    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

    bus = dbus.SystemBus()

    manager_path = find_manager(bus)
    if not manager_path:
        print('BatteryProviderManager1 interface not found')
        return

    print('BatteryProviderManager1 path = ', manager_path)

    battery_provider_manager = dbus.Interface(
            bus.get_object(BLUEZ_SERVICE_NAME, manager_path),
            BATTERY_PROVIDER_MANAGER_IFACE)

    app = Application(bus)

    # Battery pre-added before RegisterBatteryProvider
    battery1 = Battery(bus, BATTERY_PATH1, 87, 'Protocol 1')
    app.add_battery(battery1)

    mainloop = GObject.MainLoop()

    print('Registering Battery Provider...')

    battery_provider_manager.RegisterBatteryProvider(BATTERY_PROVIDER_PATH,
                                    reply_handler=register_provider_cb,
                                    error_handler=register_provider_error_cb)

    # Unregister the Battery Provider after an arbitrary amount of time
    GObject.timeout_add(
            12000, unregister_battery_provider, battery_provider_manager)
    # Simulate battery removal by a provider
    GObject.timeout_add(8000, remove_battery, app, battery1)

    mainloop.run()


if __name__ == '__main__':
    main()