mirror of
https://git.kernel.org/pub/scm/bluetooth/bluez.git
synced 2024-11-16 08:44:38 +08:00
test: Add example-endpoint
This adds an example of registering an endpoint with use of RegisterApplication.
This commit is contained in:
Luiz Augusto von Dentz
parent
01f8fc2997
commit
9d54c29375
186
test/example-endpoint
Normal file
186
test/example-endpoint
Normal file
@ -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()
|
||||
Loading…
Reference in New Issue
Block a user