From 9d54c29375bb5d3de3401e2bbadd131a5397df19 Mon Sep 17 00:00:00 2001 From: Luiz Augusto von Dentz Date: Fri, 12 Jul 2019 15:24:16 +0300 Subject: test: Add example-endpoint This adds an example of registering an endpoint with use of RegisterApplication. --- test/example-endpoint | 186 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 186 insertions(+) create mode 100644 test/example-endpoint (limited to 'test') diff --git a/test/example-endpoint b/test/example-endpoint new file mode 100644 index 000000000..a5f0348a0 --- /dev/null +++ b/test/example-endpoint @@ -0,0 +1,186 @@ +#!/usr/bin/python + +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() -- cgit v1.2.1