test/example-endpoint

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

from __future__ import absolute_import, print_function, unicode_literals

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

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

ENDPOINT_IFACE =     'org.bluez.MediaEndpoint1'
DBUS_OM_IFACE =      'org.freedesktop.DBus.ObjectManager'
DBUS_PROP_IFACE =    'org.freedesktop.DBus.Properties'

A2DP_SOURCE_UUID =   '0000110A-0000-1000-8000-00805F9B34FB'
A2DP_SINK_UUID =     '0000110B-0000-1000-8000-00805F9B34FB'

SBC_CODEC = dbus.Byte(0x00)
#Channel Modes: Mono DualChannel Stereo JointStereo
#Frequencies: 16Khz 32Khz 44.1Khz 48Khz
#Subbands: 4 8
#Blocks: 4 8 12 16
#Bitpool Range: 2-64
SBC_CAPABILITIES = dbus.Array([dbus.Byte(0xff), dbus.Byte(0xff), dbus.Byte(2), dbus.Byte(64)])
# JointStereo 44.1Khz Subbands: Blocks: 16 Bitpool Range: 2-32
SBC_CONFIGURATION = dbus.Array([dbus.Byte(0x21), dbus.Byte(0x15), dbus.Byte(2), dbus.Byte(32)])

MP3_CODEC = dbus.Byte(0x01)
#Channel Modes: Mono DualChannel Stereo JointStereo
#Frequencies: 32Khz 44.1Khz 48Khz
#CRC: YES
#Layer: 3
#Bit Rate: All except Free format
#VBR: Yes
#Payload Format: RFC-2250
MP3_CAPABILITIES = dbus.Array([dbus.Byte(0x3f), dbus.Byte(0x07), dbus.Byte(0xff), dbus.Byte(0xfe)])
# JointStereo 44.1Khz Layer: 3 Bit Rate: VBR Format: RFC-2250
MP3_CONFIGURATION = dbus.Array([dbus.Byte(0x21), dbus.Byte(0x02), dbus.Byte(0x00), dbus.Byte(0x80)])

PCM_CODEC = dbus.Byte(0x00)
PCM_CONFIGURATION = dbus.Array([], signature="ay")

CVSD_CODEC = dbus.Byte(0x01)

class Rejected(dbus.DBusException):
    _dbus_error_name = "org.bluez.Error.Rejected"

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

class Endpoint(dbus.service.Object):
    def __init__(self, bus, path, properties, configuration):
        self.path = path
        self.bus = bus
        self.properties = properties
        self.configuration = configuration
        self.exit_on_release = True
        dbus.service.Object.__init__(self, bus, self.path)

    def get_properties(self):
        return self.properties

    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 != ENDPOINT_IFACE:
            raise InvalidArgsException()

        return self.get_properties()

    def set_exit_on_release(self, exit_on_release):
        self.exit_on_release = exit_on_release

    def default_configuration(self, configuration):
        self.configuration = configuration

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

    @dbus.service.method(ENDPOINT_IFACE, in_signature="o", out_signature="")
    def ClearConfiguration(self, transport):
        print("ClearConfiguration (%s)" % (transport))

    @dbus.service.method(ENDPOINT_IFACE, in_signature="oay", out_signature="")
    def SetConfiguration(self, transport, config):
        print("SetConfiguration (%s, %s)" % (transport, config))
        return

    @dbus.service.method(ENDPOINT_IFACE, in_signature="ay", out_signature="ay")
    def SelectConfiguration(self, caps):
        print("SelectConfiguration (%s)" % (caps))
        return self.configuration

class Application(dbus.service.Object):
    def __init__(self, bus, path, properties, configuration):
        self.path = '/'
        self.endpoints = []
        dbus.service.Object.__init__(self, bus, self.path)
        self.add_endpoint(Endpoint(bus, path, properties, configuration))

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

    def add_endpoint(self, endpoint):
        self.endpoints.append(endpoint)

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

        for endpoint in self.endpoints:
            response[endpoint.get_path()] = { ENDPOINT_IFACE:
                                              endpoint.get_properties() }

        return response

def register_app_cb():
    print('Media 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()

    if len(sys.argv) > 1:
            path = bluezutils.find_adapter(sys.argv[1]).object_path
    else:
            path = bluezutils.find_adapter().object_path

    media = dbus.Interface(bus.get_object("org.bluez", path),
                           "org.bluez.Media1")


    properties = dbus.Dictionary({ "UUID" : A2DP_SOURCE_UUID,
                                   "Codec" : SBC_CODEC,
                                   "DelayReporting" : True,
                                   "Capabilities" : SBC_CAPABILITIES })

    configuration = SBC_CONFIGURATION

    if len(sys.argv) > 2:
        if sys.argv[2] == "sbcsink":
            properties = dbus.Dictionary({ "UUID" : A2DP_SINK_UUID,
                                           "Codec" : SBC_CODEC,
                                           "DelayReporting" : True,
                                           "Capabilities" : SBC_CAPABILITIES })
        if sys.argv[2] == "mp3source":
            properties = dbus.Dictionary({ "UUID" : A2DP_SOURCE_UUID,
                                           "Codec" : MP3_CODEC,
                                           "Capabilities" : MP3_CAPABILITIES })
            configuration = MP3_CONFIGURATION
        if sys.argv[2] == "mp3sink":
            properties = dbus.Dictionary({ "UUID" : A2DP_SINK_UUID,
                                           "Codec" : MP3_CODEC,
                                           "Capabilities" : MP3_CAPABILITIES })
            configuration = MP3_CONFIGURATION

    print(properties)

    path = "/test/endpoint"
    app = Application(bus, path, properties, configuration)
    mainloop = GObject.MainLoop()

    media.RegisterApplication(app.get_path(), {},
                                reply_handler=register_app_cb,
                                error_handler=register_app_error_cb)
    mainloop.run()