From: Luiz Augusto von Dentz <luiz.von.dentz@xxxxxxxxx>
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
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()
--
2.21.0
